我們以MapFunction為例，來分析下這些xxxFunction的規律

可以看到new MapFunction后面的

另外注意，這里在這里敲入new MapFunction的時候，必須在等號左側已經書寫了DataStream<Tuple2<String,Integer>>result

否則會導致intellij在MapFunction中動態補充的代碼中出現

MapFunction<String, Object>()?

而不是MapFunction<String, Tuple2<String, Integer>>()?

?

完整代碼如下:

import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.tuple.Tuple; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.typeutils.runtime.kryo.JavaSerializer; import org.apache.flink.streaming.api.TimeCharacteristic; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks; import org.apache.flink.streaming.api.functions.windowing.WindowFunction; import org.apache.flink.streaming.api.watermark.Watermark; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.Window; import org.apache.flink.util.Collector; import org.apache.flink.api.common.functions.ReduceFunction;import java.io.File;public class WindowReduce {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);DataStream<String> dataStream = env.socketTextStream("Desktop", 9999);// map的作用是 //返回的是什么，輸入的是什么DataStream<Tuple2<String,Integer>>result=dataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String s) throws Exception {return new Tuple2<String, Integer>(s,1) ;}});}}

?

總結

以上是生活随笔為你收集整理的Flink中MapFunction以及其他各种xxxFunction在intellij中的用法和规律的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。