Flink用Lambda表达式报错

Flink用Lambda表达式报错

Caused by: org.apache.flink.api.common.functions.InvalidTypesException: The generic type parameters of 'Collector' are missing. In many cases lambda methods don't provide enough information for automatic type extraction when Java generics are involved. An easy workaround is to use an (anonymous) class instead that implements the 'org.apache.flink.api.common.functions.FlatMapFunction' interface. Otherwise the type has to be specified explicitly using type information.

大致意思是，lambda写法无法提供足够的类型信息，无法推断出正确的类型，建议要么改成匿名类写法，要么用type information提供明细的类型信息。

而且官网是也是有说明的，Java Lambda Expressions | Apache Flink，于是参考了仿写了一下。

/**
 * @author WGR
 * @create 2021/7/31 -- 22:00
 */
public class WordCount {

    public static void main(String[] args)  throws Exception{
        //1.准备环境
        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();

        //2.准备数据
        DataSet<String> dataSet = environment.fromElements("dalianpai hello flink", "dalianpai hello kafka", "dalianpai");

        //3.处理数据
        DataSet<String> words = dataSet.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //value表示每一行数据
                String[] arr = value.split(" ");
                for (String word : arr) {
                    out.collect(word);
                }
            }
        });

        DataSet<Tuple2<String, Integer>> wordAndOne = words.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                //value就是每一个单词
                return Tuple2.of(value, 1);
            }
        });

        //分组
        UnsortedGrouping<Tuple2<String, Integer>> grouped = wordAndOne.groupBy(0);

        //聚合
        AggregateOperator<Tuple2<String, Integer>> result = grouped.sum(1);

        //TODO 3.sink
        result.print();
    }
}

lambda表达式:

/**
 * @author WGR
 * @create 2021/7/31 -- 22:23
 */
public class WordCount2 {

    public static void main(String[] args)  throws Exception {
        ExecutionEnvironment.getExecutionEnvironment()
                .fromElements("dalianpai hello flink", "dalianpai hello kafka", "dalianpai")
                .flatMap((FlatMapFunction<String, String>)(value, out) -> {
                    //value表示每一行数据
                    String[] arr = value.split(" ");
                    for (String word : arr) {
                        out.collect(word);
                    }
                }).returns(Types.STRING)
                .map( i -> Tuple2.of(i, 1)).returns(Types.TUPLE(Types.STRING, Types.INT))
                .groupBy(0)
                .sum(1)
                .print();
    }
}