Apache Flink：由于类型擦除，无法自动确定函数的返回类型

Question

我使用Java中的Flink编写了一个简单程序，该程序将文件或文本作为输入，然后使用flatMap函数打印所有单词。这是我的代码：        final ParameterTool params = ParameterTool.fromArgs(args);        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();        env.getConfig().setGlobalJobParameters(params);        // show user defined parameters in the apache flink dashboard        DataStream<String> dataStream;        if(params.has("input"))         {            System.out.println("Executing Words example with file input");            dataStream = env.readTextFile(params.get("input"));        }else if (params.has("host") && params.has("port"))         {            System.out.println("Executing Words example with socket stream");            dataStream = env.socketTextStream(params.get("host"), Integer.parseInt(params.get("port")));        }        else {            System.exit(1);            return;        }        DataStream<String> wordDataStream = dataStream.flatMap(                (String sentence, Collector<String> out) -> {                    for(String word: sentence.split(" "))                        out.collect(word);        });        wordDataStream.print();        env.execute("Word Split");  但是当我使用以下命令运行它时：bin/flink run -c Words FlinkExample-0.0.1-SNAPSHOT.jar --host localhost --port 9999我收到以下错误：该程序失败，但有以下异常：由于类型擦除，无法自动确定函数“ main（Words.java:32）”的返回类型。您可以通过在转换调用的结果上使用return（...）方法或通过让函数实现“ ResultTypeQueryable”接口来提供类型信息提示。（第32行指的是第二个DataStream的声明）

Answer 1

我认为错误消息的简短描述非常好，但让我对其进行扩展。

为了执行程序，Flink需要知道要处理的值的类型，因为它需要序列化和反序列化它们。Flink的类型系统基于TypeInformation它来描述数据类型。当您指定一个函数时，Flink会尝试推断该函数的返回类型。如果您的示例使用FlatMapFunction，则传递给的对象的类型Collector。

不幸的是，某些Lambda函数由于类型擦除而丢失了此信息，因此Flink无法自动推断类型。因此，您必须显式声明返回类型。

您可以提供TypeInformation，如下所示：

DataStream<String> wordDataStream = dataStream.flatMap(

    (String sentence, Collector<String> out) -> {

        for(String word: sentence.split(" "))

        out.collect(word); // collect objects of type String

    }

).returns(Types.STRING); // declare return type of flatmap lambda function as String