Flink操练（三十九）之水位线窗口

　　主要代码// 默认每隔200ms的机器时间，插入一次水位线                 .assignTimestampsAndWatermarks(                         // 最大延迟时间设置为5秒                         WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(5))                                 .withTimestampAssigner(new SerializableTimestampAssigner>() {                                     @Override                                     public long extractTimestamp(Tuple2 element, long recordTimestamp) {                                         return element.f1; // 告诉flink事件时间是哪一个字段                                     }                                 })                 )                 .keyBy(r -> r.f0)                 .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 5秒的事件时间滚动窗口                 .process(new ProcessWindowFunction, String, String, TimeWindow>() {                     @Override                     public void process(String key, Context context, Iterable> elements, Collector out) throws Exception {                         long windowStart = context.window().getStart();                         long windowEnd   = context.window().getEnd();                         long count       = elements.spliterator().getExactSizeIfKnown(); // 迭代器里面共多少条元素                         out.collect(＂用户：＂ + key + ＂ 在窗口＂ +                                 ＂＂ + new Timestamp(windowStart) + ＂~＂ + new Timestamp(windowEnd) + ＂＂ +                                 ＂中的pv次数是：＂ + count);                     }                 })                 .print();完整代码package day04;  import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSink; import org.apache.flink.streaming.api.datastream.DataStreamSource; 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 java.sql.Timestamp; import java.time.Duration;  /**  * @program: bigData_learn  * @description: 水位线测试  * @author: Mr.逗  * @create: 2021-09-23 18:01  **/ public class WaterTest {     public static void main(String[] args) {         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         env.setParallelism(1);         // `a 1`         DataStreamSource source = env.socketTextStream(＂172.17.0.50＂, 9999);         // (a, 1000L)         DataStreamSink map = source.map(new MapFunction>() {             @Override             public Tuple2 map(String value) throws Exception {                 String[] arr = value.split(＂ ＂);                 return Tuple2.of(arr[0], Long.parseLong(arr[1]) * 1000L);             }         })                 // 默认每隔200ms的机器时间，插入一次水位线                 .assignTimestampsAndWatermarks(                         // 最大延迟时间设置为5秒                         WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(5))                                 .withTimestampAssigner(new SerializableTimestampAssigner>() {                                     @Override                                     public long extractTimestamp(Tuple2 element, long recordTimestamp) {                                         return element.f1; // 告诉flink事件时间是哪一个字段                                     }                                 })                 )                 .keyBy(r -> r.f0)                 .window(TumblingEventTimeWindows.of(Time.seconds(5))) // 5秒的事件时间滚动窗口                 .process(new ProcessWindowFunction, String, String, TimeWindow>() {                     @Override                     public void process(String key, Context context, Iterable> elements, Collector out) throws Exception {                         long windowStart = context.window().getStart();                         long windowEnd   = context.window().getEnd();                         long count       = elements.spliterator().getExactSizeIfKnown(); // 迭代器里面共多少条元素                         out.collect(＂用户：＂ + key + ＂ 在窗口＂ +                                 ＂＂ + new Timestamp(windowStart) + ＂~＂ + new Timestamp(windowEnd) + ＂＂ +                                 ＂中的pv次数是：＂ + count);                     }                 })                 .print();         String name = WaterTest.class.getName();         try {             env.execute(name);         }catch (Exception e)         {             e.printStackTrace();         }     } }