flink 问题记录

 转发请注明原创地址：http://www.cnblogs.com/dongxiao-yang/p/7652337.html

1 WindowFunction类型不匹配无法编译。

flink 版本：1.3.0

参考https://ci.apache.org/projects/flink/flink-docs-release-1.3/dev/windows.html#windowfunction-with-incremental-aggregation写的demo发现reduce加入MyWindowFunction后编译不通过，报错参数类型不匹配。

代码如下

        MyReduceFunction a = new MyReduceFunction();
        
        

        DataStream<Tuple3<String, String, Integer>> counts4 = source

                .keyBy(0)
                .window(TumblingEventTimeWindows.of(Time
                        .of(1, TimeUnit.SECONDS)))
                .reduce(a,
                        new WindowFunction<Tuple2<String, Integer>, Tuple3<String, String, Integer>, String, TimeWindow>() {
                            private static final long serialVersionUID = 1L;

                            @Override
                            public void apply(
                                    String key,
                                    TimeWindow window,
                                    Iterable<Tuple2<String, Integer>> values,
                                    Collector<Tuple3<String, String, Integer>> out)
                                    throws Exception {
                                for (Tuple2<String, Integer> in : values) {
                                    out.collect(new Tuple3<>(in.f0, in.f0,
                                            in.f1));
                                }
                            }
                        });    



public static class MyReduceFunction implements
            ReduceFunction<Tuple2<String, Integer>> {

        @Override
        public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1,
                Tuple2<String, Integer> value2) throws Exception {
            // TODO Auto-generated method stub
            return new Tuple2<String, Integer>(value1.f0,
                    (value1.f1 + value2.f1));
        }

    }

View Code