一、ProcessFunction的使用

1、沒有進行keyBy

package cn._51doit.flink.day07;import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.util.Collector;public class NonKeyedProcessFunction {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);//對NonKeyedDataStream調用ProcessSingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = lines.process(new ProcessFunction<String, Tuple2<String, Integer>>() {@Overridepublic void processElement(String line, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {String[] words = line.split(" ");for (String word : words) {if (!word.equals("error")) {out.collect(Tuple2.of(word, 1));}}}});wordAndOne.print();env.execute();} }

2、有keyBy

package cn._51doit.flink.day07;import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.restartstrategy.RestartStrategies; import org.apache.flink.api.common.state.MapState; import org.apache.flink.api.common.state.MapStateDescriptor; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; import org.apache.flink.util.Collector;/****/ public class KeyedProcessFunctionDemo {public static void main(String[] args) throws Exception{//創建Flink流計算執行環境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(10000);//設置重啟策略env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));//創建DataStream//SourceDataStreamSource<String> lines = env.socketTextStream("localhost", 8888);//調用Transformation開始//調用TransformationSingleOutputStreamOperator<Tuple3<String, String, Double>> tpDataStream = lines.map(new MapFunction<String, Tuple3<String, String, Double>>() {@Overridepublic Tuple3<String, String, Double> map(String value) throws Exception {String[] fields = value.split(",");return Tuple3.of(fields[0], fields[1], Double.parseDouble(fields[2]));}});KeyedStream<Tuple3<String, String, Double>, String> keyedStream = tpDataStream.keyBy(t -> t.f0);//對KeyedDataStream調用process方法，可以獲取KeyedStateSingleOutputStreamOperator<Tuple3<String, String, Double>> result = keyedStream.process(new KeyedProcessFunction<String, Tuple3<String, String, Double>, Tuple3<String, String, Double>>() {private transient MapState<String, Double> mapState;@Overridepublic void open(Configuration parameters) throws Exception {//定義一個狀態描述器MapStateDescriptor<String, Double> stateDescriptor = new MapStateDescriptor<String, Double>("kv-state", String.class, Double.class);//初始化或恢復歷史狀態mapState = getRuntimeContext().getMapState(stateDescriptor);}@Overridepublic void processElement(Tuple3<String, String, Double> value, Context ctx, Collector<Tuple3<String, String, Double>> out) throws Exception {String city = value.f1;Double money = value.f2;Double historyMoney = mapState.get(city);if (historyMoney == null) {historyMoney = 0.0;}Double totalMoney = historyMoney + money; //累加//更新到state中mapState.put(city, totalMoney);//輸出value.f2 = totalMoney;out.collect(value);}});result.print();//啟動執行env.execute("StreamingWordCount");} }

二、定時器

1、基本使用

package cn._51doit.flink.day07;import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; import org.apache.flink.util.Collector;public class ProcessingTimeTimerDemo {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//spark,1//spark,2//hadoop,1DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String line) throws Exception {String[] fields = line.split(",");return Tuple2.of(fields[0], Integer.parseInt(fields[1]));}});KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);keyed.process(new KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String, Integer>>() {@Overridepublic void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {//獲取當前的ProcessingTimelong currentProcessingTime = ctx.timerService().currentProcessingTime();//System.out.println("當前時間：" + currentProcessingTime + "，定時器觸發的時間：" + (currentProcessingTime + 30000));//將當前的ProcessingTime + 30 秒，注冊一個定時器ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 30000);}//當鬧鐘到了指定的時間，就執行onTimer方法@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {System.out.println("定時器執行了：" + timestamp);}}).print();env.execute();} }

2、先把數據攢起來，滿足條件了再輸出

package cn._51doit.flink.day07;import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.KeyedProcessFunction; import org.apache.flink.util.Collector;public class ProcessingTimeTimerDemo02 {public static void main(String[] args) throws Exception{StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();//spark,1//spark,2//hadoop,1DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String line) throws Exception {String[] fields = line.split(",");return Tuple2.of(fields[0], Integer.parseInt(fields[1]));}});KeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);keyed.process(new KeyedProcessFunction<String, Tuple2<String, Integer>, Tuple2<String, Integer>>() {private transient ValueState<Integer> counter;@Overridepublic void open(Configuration parameters) throws Exception {ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("wc-state", Integer.class);counter = getRuntimeContext().getState(stateDescriptor);}@Overridepublic void processElement(Tuple2<String, Integer> value, Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {//獲取當前的ProcessingTimelong currentProcessingTime = ctx.timerService().currentProcessingTime();long fireTime = currentProcessingTime - currentProcessingTime % 60000 + 60000;//下一分鐘//如果注冊相同數據的TimeTimer，后面的會將前面的覆蓋，即相同的timeTimer只會觸發一次ctx.timerService().registerProcessingTimeTimer(fireTime);Integer currentCount = value.f1;Integer historyCount = counter.value();if(historyCount == null) {historyCount = 0;}Integer totalCount = historyCount + currentCount;//更新狀態counter.update(totalCount);}//當鬧鐘到了指定的時間，就執行onTimer方法@Overridepublic void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<String, Integer>> out) throws Exception {//定時器觸發，輸出當前的結果Integer value = counter.value();String currentKey = ctx.getCurrentKey();//輸出key，Value//如果想要實現類似滾動窗口，不累加類似數據，只是累加當前窗口的數據，就清空狀態//counter.update(0);out.collect(Tuple2.of(currentKey, value));}}).print();env.execute();} }

三、測流輸出 / 旁路輸出

1、獲取不同類型的數據，打上不同的標簽

package cn._51doit.flink.day07;import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.util.Collector; import org.apache.flink.util.OutputTag;public class SideOutputDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);//奇數OutputTag<String> oddOutputTag = new OutputTag<String>("odd") {};//偶數OutputTag<String> evenOutputTag = new OutputTag<String>("even") {};//非數字OutputTag<String> nanOutputTag = new OutputTag<String>("nan") {};SingleOutputStreamOperator<String> mainStream = lines.process(new ProcessFunction<String, String>() {@Overridepublic void processElement(String value, Context ctx, Collector<String> out) throws Exception {try {int i = Integer.parseInt(value);if (i % 2 == 0) {//偶數ctx.output(evenOutputTag, value);} else {//奇數ctx.output(oddOutputTag, value);}} catch (NumberFormatException e) {ctx.output(nanOutputTag, value);}//在主流中輸出全部的數據out.collect(value);}});//偶數DataStream<String> evenStream = mainStream.getSideOutput(evenOutputTag);//奇數DataStream<String> oddStream = mainStream.getSideOutput(oddOutputTag);oddStream.print("odd: ");evenStream.print("even: ");mainStream.print("main: ");env.execute();} }

2、使用側流輸出獲取窗口遲到的數據

package cn._51doit.flink.day07;import org.apache.flink.api.common.eventtime.IngestionTimeAssigner; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.*; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.functions.windowing.WindowFunction; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.util.Collector; import org.apache.flink.util.OutputTag;import java.time.Duration;//使用側流輸出獲取窗口遲到的數據 public class WindowLateDateDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());//1000,spark,3//1200,spark,5//2000,hadoop,2//socketTextStream返回的DataStream并行度為1DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<String> dataWithWaterMark = lines.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner((line, timestamp) -> Long.parseLong(line.split(",")[0])));SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {String[] fields = value.split(",");return Tuple2.of(fields[1], Integer.parseInt(fields[2]));}});//調用keyByKeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);OutputTag<Tuple2<String, Integer>> lateDataTag = new OutputTag<Tuple2<String, Integer>>("late-data") {};//NonKeyd Window： 不調用KeyBy，然后調用windowAll方法，傳入windowAssinger// Keyd Window： 先調用KeyBy，然后調用window方法，傳入windowAssingerWindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));.sideOutputLataData(lateDataTag); //將遲到數據打上標簽SingleOutputStreamOperator<Tuple2<String, Integer>> summed = windowed.sum(1);summed.print();DataStream<Tuple2<String, Integer>> lataDataStream=summed.getSideOutput(lateDataTag); //從主流當中獲取遲到數據lataDataStream.print("lata-Data: ");env.execute();}}

四、WindowFunction 使用

1、窗口內增量聚合，且與歷史數據聚合

package cn._51oit.flink.day07;import org.apache.flink.api.common.eventtime.IngestionTimeAssigner; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.*; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.functions.windowing.WindowFunction; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.util.Collector; import org.apache.flink.util.OutputTag;import java.time.Duration;//使用側流輸出獲取窗口遲到的數據 public class WindowLateDateDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());//1000,spark,3//1200,spark,5//2000,hadoop,2//socketTextStream返回的DataStream并行度為1DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<String> dataWithWaterMark = lines.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner((line, timestamp) -> Long.parseLong(line.split(",")[0])));SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {String[] fields = value.split(",");return Tuple2.of(fields[1], Integer.parseInt(fields[2]));}});//調用keyByKeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);//NonKeyd Window： 不調用KeyBy，然后調用windowAll方法，傳入windowAssinger//Keyd Window： 先調用KeyBy，然后調用window方法，傳入windowAssingerWindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));//如果直接調用sum或reduce，只會聚合窗口內的數據，不去跟歷史數據進行累加//需求：可以在窗口內進行增量聚合，并且還可以與歷史數據進行聚合SingleOutputStreamOperator<Tuple2<String, Integer>> result = windowed.reduce(new MyReduceFunc(), new MyWindowFunc());result.print();env.execute();}public static class MyReduceFunc implements ReduceFunction<Tuple2<String, Integer>> {@Overridepublic Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {value1.f1 = value1.f1 + value2.f1;return value1;}}public static class MyWindowFunc extends ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow> {private transient ValueState<Integer> sumState;@Overridepublic void open(Configuration parameters) throws Exception {ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("wc-state", Integer.class);sumState = getRuntimeContext().getState(stateDescriptor);}@Overridepublic void process(String key, Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {Integer historyCount = sumState.value();if (historyCount == null) {historyCount = 0;}//獲取到窗口聚合后輸出的結果Tuple2<String, Integer> tp = elements.iterator().next();Integer windowCount = tp.f1;Integer totalCount = historyCount + windowCount;//更新狀態sumState.update(totalCount);tp.f1 = totalCount;//輸出out.collect(tp);}} }

2、aggregate結合WindowFunction

package cn._51doit.flink.day07;import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.AggregateFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.datastream.WindowedStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.util.Collector; import org.apache.flink.util.OutputTag;import java.time.Duration;//累加當前窗口的數據，并與歷史數據進行累加 public class WindowAggregateFunctionDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());env.enableCheckpointing(10000);//1000,spark,3//1200,spark,5//2000,hadoop,2//socketTextStream返回的DataStream并行度為1DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<String> dataWithWaterMark = lines.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner((line, timestamp) -> Long.parseLong(line.split(",")[0])));SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {String[] fields = value.split(",");return Tuple2.of(fields[1], Integer.parseInt(fields[2]));}});//調用keyByKeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);OutputTag<Tuple2<String, Integer>> lateDataTag = new OutputTag<Tuple2<String, Integer>>("late-data") {};//NonKeyd Window： 不調用KeyBy，然后調用windowAll方法，傳入windowAssinger// Keyd Window： 先調用KeyBy，然后調用window方法，傳入windowAssingerWindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));//如果直接調用sum或reduce，只會聚合窗口內的數據，不去跟歷史數據進行累加//需求：可以在窗口內進行增量聚合，并且還可以與歷史數據進行聚合SingleOutputStreamOperator<Tuple2<String, Integer>> result = windowed.aggregate(new MyAggFunc(), new MyWindowFunc());result.print();env.execute();}private static class MyAggFunc implements AggregateFunction<Tuple2<String, Integer>, Integer, Integer> {//創建一個初始值@Overridepublic Integer createAccumulator() {return 0;}//數據一條數據，與初始值或中間累加的結果進行聚合@Overridepublic Integer add(Tuple2<String, Integer> value, Integer accumulator) {return value.f1 + accumulator;}//返回的結果@Overridepublic Integer getResult(Integer accumulator) {return accumulator;}//如果使用的是非SessionWindow，可以不實現@Overridepublic Integer merge(Integer a, Integer b) {return null;}}private static class MyWindowFunc extends ProcessWindowFunction<Integer, Tuple2<String, Integer>, String, TimeWindow> {private transient ValueState<Integer> sumState;@Overridepublic void open(Configuration parameters) throws Exception {ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("wc-state", Integer.class);sumState = getRuntimeContext().getState(stateDescriptor);}@Overridepublic void process(String key, Context context, Iterable<Integer> elements, Collector<Tuple2<String, Integer>> out) throws Exception {Integer historyCount = sumState.value();if (historyCount == null) {historyCount = 0;}//獲取到窗口聚合后輸出的結果Integer windowCount = elements.iterator().next();Integer totalCount = historyCount + windowCount;//更新狀態sumState.update(totalCount);//輸出out.collect(Tuple2.of(key, totalCount));}} }

3、ProcessWindowFunction使用

package cn._51doit.flink.day07;import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.AggregateFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.functions.ReduceFunction; import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.KeyedStream; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.datastream.WindowedStream; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.util.Collector; import org.apache.flink.util.OutputTag;import java.time.Duration;//累加當前串口的數據，并與歷史數據進行累加 public class WindowProcessFunctionDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());env.enableCheckpointing(10000);//1000,spark,3//1200,spark,5//2000,hadoop,2//socketTextStream返回的DataStream并行度為1DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);SingleOutputStreamOperator<String> dataWithWaterMark = lines.assignTimestampsAndWatermarks(WatermarkStrategy.<String>forBoundedOutOfOrderness(Duration.ZERO).withTimestampAssigner((line, timestamp) -> Long.parseLong(line.split(",")[0])));SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndCount = dataWithWaterMark.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {String[] fields = value.split(",");return Tuple2.of(fields[1], Integer.parseInt(fields[2]));}});//調用keyByKeyedStream<Tuple2<String, Integer>, String> keyed = wordAndCount.keyBy(t -> t.f0);//NonKeyd Window： 不調用KeyBy，然后調用windowAll方法，傳入windowAssinger// Keyd Window： 先調用KeyBy，然后調用window方法，傳入windowAssingerWindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed.window(TumblingEventTimeWindows.of(Time.seconds(5)));//如果直接調用sum或reduce，只會聚合窗口內的數據，不去跟歷史數據進行累加//需求：可以在窗口內進行增量聚合，并且還可以與歷史數據進行聚合SingleOutputStreamOperator<Tuple2<String, Integer>> result = windowed.process(new ProcessWindowFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, String, TimeWindow>() {//窗口觸發，才會調用process方法，該方法可以獲取窗口內的全量獲取窗口的數據，數據是緩存到windowstate中的@Overridepublic void process(String s, Context context, Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {for (Tuple2<String, Integer> element : elements) {out.collect(element);}}});result.print();env.execute();}}

總結

以上是生活随笔為你收集整理的大数据之flink定时器的全部內容，希望文章能夠幫你解決所遇到的問題。

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