Flink executes the program and reports an error:
/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/bin/java -javaagent:/Applications/IntelliJ IDEA.app/Contents/lib/idea_rt.jar=51984:/Applications/IntelliJ IDEA.app/Contents/bin -Dfile.encoding=UTF-8 -classpath /Libr ary/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/charsets.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/deploy.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_ 301.jdk/Contents/Home/jre/lib/ext/cldrdata.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/ext/dnsns.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre /lib/ext/jaccess.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/ext/jfxrt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/ext/localedata.jar:/Library /Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/ext/nashorn.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/ext/sunec.jar:/Library/Java/JavaVirtualMachines/jdk1.8 .0_301.jdk/Contents/Home/jre/lib/ext/sunjce_provider.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/ext/sunpkcs11.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.j dk/Contents/Home/jre/lib/ext/zipfs.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/javaws.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/jce. jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/jfr.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/jfxswt.jar:/Library/Java/JavaVirtualMachines/j dk1.8.0_301.jdk/Contents/Home/jre/lib/jsse.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/management-agent.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents /Home/jre/lib/plugin.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/resources.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/jre/lib/rt.jar:/Library/Java/ JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/lib/ant-javafx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/lib/dt.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Con contents/Home/lib/javafx-mx.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/lib/jconsole.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/lib/packager.jar:/Library/Java/Java VirtualMachines/jdk1.8.0_301.jdk/Contents/Home/lib/sa-jdi.jar:/Library/Java/JavaVirtualMachines/jdk1.8.0_301.jdk/Contents/Home/lib/tools.jar:/Users/jason93/IdeaProjects/Hadoop/Flink/target/classes:/Users/ jason93/.m2/repository/org/apache/flink/flink-java/1.13.6/flink-java-1.13.6.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-core/1.13.6/flink-core-1.13.6.jar:/Users/jason93/.m2/repository /org/apache/flink/flink-annotations/1.13.6/flink-annotations-1.13.6.jar:/Users/jason93/.m2/repository /org/apache/flink/flink-metrics-core/1.13.6/flink-metrics-core-1.13.6.jar:/Users/jason93/.m2/repository story/org/apache/flink/flink-shaded-asm-7/7.1-13.0/flink-shaded-asm-7-7.1-13.0.jar:/Users/jason93/.m2/repository/com/esotericsoftware/kryo/kryo/2.24.0/kryo-2.24.0.jar:/Users/jason93/.m 2/repository/com/esotericsoftware/minlog/minlog/1.2/minlog-1.2.jar:/Users/jason93/.m2/repository/org/objenesis/objenesis/2.1/objenesis-2.1.jar:/Users/jason93/.m2/repository/commons-collections/commons-collections/3. 2.2/commons-collections-3.2.2.jar:/Users/jason93/.m2/repository/org/apache/commons/commons-compress/1.21/commons-compress-1.21.jar:/Users/jason93/.m2/repository/org/apache/commons/commons-lang3/3.3.2/commons- lang3-3.3.2.jar:/Users/jason93/.m2/repository/org/apache/commons/commons-math3/3.5/commons-math3-3.5.jar:/Users/jason93/.m2/repository/com/google/code/findbugs/jsr305/1.3.9/jsr305-1.3.9. jar:/Users/jason93/.m2/repository/org/apache/flink/force-shading/1.13.6/force-shading-1.13.6.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-streaming-java_2.12/1.13.6/flink-streaming-java_2.12- 1.13.6.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-file-sink-common/1.13.6/flink-file-sink-common-1.13.6.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-runtime_2.12/1.13.6/flink -runtime_2.12-1.13.6.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-queryable-state-client-java/1.13.6/flink-queryable-state-client-java-1.13.6.jar:/Users/jason93/.m2/repository/org/apache/flink/f link-hadoop-fs/1.13.6/flink-hadoop-fs-1.13.6.jar:/Users/jason93/.m2/repository/commons-io/commons-io/2.8.0/commons-io-2.8.0.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-shaded-netty /4.1.49.Final-13.0/flink-shaded-netty-4.1.49.Final-13.0.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-shaded-jackson/2.12.1-13.0/flink-shaded-jackson-2.12.1-13.0.jar:/Users /jason93/.m2/repository/org/apache/flink/flink-shaded-zookeeper-3/3.4.14-13.0/flink-shaded-zookeeper-3-3.4.14-13.0.jar:/Users/jason93/.m2/repository/org/javassist/javassist/3.24.0-GA/javassist- 3.24.0-GA.jar:/Users/jason93/.m2/repository/com/typesafe/akka/akka-actor_2.12/2.5.21/akka-actor_2.12-2.5.21.jar:/Users/jason93/.m2/repository/com/typesafe/config/1.3.3/config-1.3.3.jar:/User s/jason93/.m2/repository/org/scala-lang/modules/scala-java8-compat_2.12/0.8.0/scala-java8-compat_2.12-0.8.0.jar:/Users/jason93/.m2/repository/com/typesafe/akka/akka-stream_2.12/2.5.21/akka-stream _2.12-2.5.21.jar:/Users/jason93/.m2/repository/org/reactivestreams/reactive-streams/1.0.2/reactive-streams-1.0.2.jar:/Users/jason93/.m2/repository/com/typesafe/ssl-config-core_2.12/0.3.7/ssl-config-co re_2.12-0.3.7.jar:/Users/jason93/.m2/repository/org/scala-lang/modules/scala-parser-combinators_2.12/1.1.1/scala-parser-combinators_2.12-1.1.1.jar:/Users/jason93/.m2/repository/com/typesafe/akka/akka- protobuf_2.12/2.5.21/akka-protobuf_2.12-2.5.21.jar:/Users/jason93/.m2/repository/com/typesafe/akka/akka-slf4j_2.12/2.5.21/akka-slf4j_2.12-2.5.21.jar:/Users/jason93/.m2/ repository/org/clapper/grizzled-slf4j_2.12/1.3.2/grizzled-slf4j_2.12-1.3.2.jar:/Users/jason93/.m2/repository/com/github/scopt/scopt_2.12/3.5.0/scopt_2.12-3.5.0.jar:/Users/jason 93/.m2/repository/org/xerial/snappy/snappy-java/1.1.8.3/snappy-java-1.1.8.3.jar:/Users/jason93/.m2/repository/com/twitter/chill_2.12/0.7.6/chill_2.12-0.7.6.jar:/Users/jason93/.m 2/repository/com/twitter/chill-java/0.7.6/chill-java-0.7.6.jar:/Users/jason93/.m2/repository/org/lz4/lz4-java/1.6.0/lz4-java-1.6.0.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-shaded -guava/18.0-13.0/flink-shaded-guava-18.0-13.0.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-streaming-scala_2.12/1.13.6/flink-streaming-scala_2.12-1.13.6.jar:/Users/jason93 /.m2/repository/org/apache/flink/flink-scala_2.12/1.13.6/flink-scala_2.12-1.13.6.jar:/Users/jason93/.m2/repository/org/scala-lang/scala-reflect/2.12.7/scala-reflect-2.12.7.jar:/Users/jason93 /.m2/repository/org/scala-lang/scala-library/2.12.7/scala-library-2.12.7.jar:/Users/jason93/.m2/repository/org/scala-lang/scala-compiler/2.12.7/scala-compiler-2.12.7.jar:/Users/jason93/.m2/repository /org/scala-lang/modules/scala-xml_2.12/1.0.6/scala-xml_2.12-1.0.6.jar:/Users/jason93/.m2/repository/org/apache/flink/flink-clients_2.12/1.13.6/flink-clients_2.12-1.13.6.jar:/Users/jason9 3/.m2/repository/org/apache/flink/flink-optimizer_2.12/1.13.6/flink-optimizer_2.12-1.13.6.jar:/Users/jason93/.m2/repository/commons-cli/commons-cli/1.3.1/commons-cli-1.3.1.jar:/Users/ jason93/.m2/repository/org/slf4j/slf4j-api/1.7.30/slf4j-api-1.7.30.jar:/Users/jason93/.m2/repository/org/slf4j/slf4j-log4j12/1.7.30/slf4j-log4j12-1.7.30.jar:/Users/jason9 3/.m2/repository/log4j/log4j/1.2.17/log4j-1.2.17.jar:/Users/jason93/.m2/repository/org/apache/logging/log4j/log4j-to-slf4j/2.14.0/log4j-to-slf4j-2.14.0.jar:/Users/jason93/.m2 /repository/org/apache/logging/log4j/log4j-api/2.14.0/log4j-api-2.14.0.jar:/Volumes/software/BigData/Scala/scala-2.12.11/lib/scala-library.jar:/Volumes/software/BigData/Scala/scala-2.12.11/lib/scala-swing_2. 12-2.0.3.jar:/Volumes/software/BigData/Scala/scala-2.12.11/lib/scala-reflect.jar:/Volumes/software/BigData/Scala/scala-2.12.11/lib/scala-parser-combinators_2.12-1.0.7.jar:/Volumes/software/BigData/Scala/scala-2. 12.11/lib/scala-xml_2.12-1.0.6.jar WordCount (hello, 2) (scala,1) (spark,1) Exception in thread "main" org.apache.flink.api.common.functions.InvalidTypesException: The return type of function 'dataStreamWordCount(WordCount.java:85)' could not be determined automatically, due to type erasure. You can give type information hints by us ing 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:479) at org.apache.flink.streaming.api.datastream.DataStream.getType(DataStream.java:193) at org.apache.flink.streaming.api.datastream.DataStream.keyBy(DataStream.java:319) at WordCount.dataStreamWordCount(WordCount.java:92) at WordCount.main(WordCount.java:24) 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 The 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 WordCount.dataStreamWordCount(WordCount.java:85) ... 1 more Process finished with exit code 1
After analysis, referencing the Java8 Lambda form will cause generic erasure. JDK cannot recognize the data type inside the object and cannot automatically convert it. The code is:
public void dataStreamWordCount(String path) throws Exception {<!-- -->
// StreamExecutionEnvironment finally inherits DataStream
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env. setParallelism(1);
SingleOutputStreamOperator<Tuple2<String, Integer>> streamOperator =
env.readTextFile(path).flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {<!-- -->
String[] words = line. split(" ");
for (String word : words) {<!-- -->
out.collect(Tuple2.of(word, 1));
}
});
streamOperator.keyBy(0).sum(1).print();
env. execute();
The exception code is after flatMap.
Solution 1: According to the normal way, new a flatMap method to write
env.readTextFile(path).flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {<!-- -->
@Override
public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {<!-- -->
String[] words = line. split(" ");
for (String word : words) {<!-- -->
out.collect(Tuple2.of(word, 1));
}
}
});
Solution 2: Since JDK cannot automatically convert, then explicitly specify the internal structure of the object and expose it to JDK
flatMap((String line, Collector<Tuple2<String, Integer>> out) -> {<!-- -->
String[] words = line. split(" ");
for (String word : words) {<!-- -->
out.collect(Tuple2.of(word, 1));
}
}).returns(Types. TUPLE(Types. STRING, Types. INT))