Flink操练（四十一）之周期水位线窗口

　　关键代码env.getConfig().setAutoWatermarkInterval(60* 1000L);完整代码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.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-24 09:30  **/ public class WaterTest2 {     public static void main(String[] args) {         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         env.setParallelism(1);         // 每隔1分钟插入一次水位线         env.getConfig().setAutoWatermarkInterval(60* 1000L);         DataStreamSource source = env.socketTextStream(＂172.17.0.50＂, 9999);         DataStream> map = source.map(new MapFunction>() {             @Override             public Tuple2 map(String v) throws Exception {                 return Tuple2.of(v.split(＂ ＂)[0], Long.parseLong(v.split(＂ ＂)[1]) * 1000L);             }         });         // 默认每隔200ms的机器时间，插入一次水位线         DataStream> watermarks = map.assignTimestampsAndWatermarks(                 WatermarkStrategy.>forBoundedOutOfOrderness(Duration.ofSeconds(0))                         .withTimestampAssigner(new SerializableTimestampAssigner>() {                             @Override                             public long extractTimestamp(Tuple2 v, long l) {                                 return v.f1;// 告诉flink事件时间是哪一个字段                             }                         })         );         DataStream process = watermarks.keyBy(v -> v.f0)                 //5秒的事件时间滚动窗口                 .window(TumblingEventTimeWindows.of(Time.seconds(5)))                 .process(new ProcessWindowFunction, String, String, TimeWindow>() {                     @Override                     public void process(String key, Context ctx, Iterable> it, Collector out) throws Exception {                         long start = ctx.window().getStart();                         long end = ctx.window().getEnd();                         long count = it.spliterator().getExactSizeIfKnown();//迭代器里面共有多少元素                         out.collect(＂用户：＂ + key + ＂在窗口＂ +                                 ＂＂ + new Timestamp(start) + ＂~＂ + new Timestamp(end) + ＂＂ +                                 ＂中的pv次数是：＂ + count);                     }                 });         process.print();         String name = WaterTest2.class.getName();         try {             env.execute(name);         }catch (Exception e)         {             e.printStackTrace();         }     } }