Flink-Java报错之org.apache.flink.api.common.functions.InvalidTypesException

在启动Flink作业时报错。可以看到是在实现FlatMapFunction时InvalidTypesException类型无效。提示使用匿名类或者明确指定类型来解决。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.

Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException: The return type of function 'main(WordCount.java:22)' could not be determined automatically, due to type erasure. You can give type information hints by using the returns(...) method on the result of the transformation call, or by letting your function implement the 'ResultTypeQueryable' interface.
	at org.apache.flink.api.dag.Transformation.getOutputType(Transformation.java:496)
	at org.apache.flink.streaming.api.datastream.DataStream.getType(DataStream.java:193)
	at org.apache.flink.streaming.api.datastream.KeyedStream.(KeyedStream.java:116)
	at org.apache.flink.streaming.api.datastream.DataStream.keyBy(DataStream.java:293)
	at example.WordCount.main(WordCount.java:29)
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.
	at org.apache.flink.api.java.typeutils.TypeExtractionUtils.validateLambdaType(TypeExtractionUtils.java:371)
	at org.apache.flink.api.java.typeutils.TypeExtractionUtils.extractTypeFromLambda(TypeExtractionUtils.java:188)
	at org.apache.flink.api.java.typeutils.TypeExtractor.getUnaryOperatorReturnType(TypeExtractor.java:557)
	at org.apache.flink.api.java.typeutils.TypeExtractor.getFlatMapReturnTypes(TypeExtractor.java:174)
	at org.apache.flink.streaming.api.datastream.DataStream.flatMap(DataStream.java:612)
	at example.WordCount.main(WordCount.java:22)

可以搜索这个报错，有很多文章提到了解决方法，并给出了示例代码。

但是这个报错的原因是当涉及Java泛型时，lambda方法不能为自动类型提取提供足够的信息。也就是说由于类型擦除，无法自动确定。所以其实还有一个很简单的解决方案，就是不用Lambda表达式

SingleOutputStreamOperator> sum =
    source
    .flatMap(
        new FlatMapFunction>() {
            @Override
            public void flatMap(String s, Collector> collector) throws Exception {
                for (String string : s.split(" ")) {
                    collector.collect(new Tuple2<>(string,1L));
                }
           }
        }
    )
    .keyBy(s->s.f0).sum(1);
sum.print("wc:");