Flink--8、时间语义、水位线（事件和窗口、水位线和窗口的工作原理、生产水位线、水位线的传递、迟到数据的处理）

DataStream<Event> stream = env.addSource(new ClickSource());
DataStream<Event> withTimestampsAndWatermarks = 
stream.assignTimestampsAndWatermarks(<watermark strategy>);

public interface WatermarkStrategy<T> 
    extends TimestampAssignerSupplier<T>,
            WatermarkGeneratorSupplier<T>{
    // 负责从流中数据元素的某个字段中提取时间戳，并分配给元素。时间戳的分配是生成水位线的基础。
    @Override
    TimestampAssigner<T> createTimestampAssigner(TimestampAssignerSupplier.Context context);
    // 主要负责按照既定的方式，基于时间戳生成水位线
    @Override
    WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);
}

/**
 * 在中间环节产生水印，需要使用：assignTimestampsAndWatermarks(WatermarkStrategy x)
 *
 * WatermarkStrategy:水印策略。
 *          包含以下信息：
 *              （1）水印的特征
 *                  （a）无水印：watermarkStrategy.noWatermarks()
 *                  (b)自定义水印特征 WatermarkStrategy.forGenerator()
 *                  选择系统已经提供：
 *                      （c）连续水印：数据中提取时间属性-1ms-0
 *                              WatermarkStrategy.forMonotonousTimeStamps()
 *                       (d)乱序水印：数据中提前的时间属性-1ms-自定义间隔时间
 *                               WatermarkStrategy.forBoundedOutOfOrderNess()
 *              (2)水印的计算方式
 *                     水印从数据的时间属性中计算得到
 *                     计算方式的核心功能就是告诉算子，数据中的哪个属性是事件属性
 *   ----------------------------------------------------------------------------
 *   一开始玩，一定要把并行度设置为1
 */
public class Demo01_ShowWaterMark {
     public static void main(String[] args) throws Exception {
             //创建Flink配置类（空参创建的话都是默认值）
             Configuration configuration = new Configuration();
             //修改配置类中的WebUI端口号
             configuration.setInteger("rest.port",3333);
             //创建Flink环境（并且传入配置对象）
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
             //不合并操作算子
         env.disableOperatorChaining();
         //设置并行度是1
         env.setParallelism(1);
         //自定义水印策略
         WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy.<WaterSensor>forMonotonousTimestamps()
                 .withTimestampAssigner((e, ts) -> e.getTs());
         env.socketTextStream("hadoop102",9999)
                 .map(new WaterSensorFunction())
                 .assignTimestampsAndWatermarks(watermarkStrategy)
                 .keyBy(WaterSensor::getId)
                 .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                     @Override
                     public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                         out.collect(value+"="+ctx.timerService().currentWatermark());
                     }
                 })
                 .print();
         env.execute();
     }
}

public class WatermarkOutOfOrdernessDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("hadoop102", 7777)
                .map(new WaterSensorMapFunction());
        // TODO 1.定义Watermark策略
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
                // 1.1 指定watermark生成：乱序的，等待3s
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                // 1.2 指定 时间戳分配器，从数据中提取
                .withTimestampAssigner(
                        (element, recordTimestamp) -> {
                            // 返回的时间戳，要 毫秒
                            System.out.println("数据=" + element + ",recordTs=" + recordTimestamp);
                            return element.getTs() * 1000L;
                        });
        // TODO 2. 指定 watermark策略
        SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
        sensorDSwithWatermark.keyBy(sensor -> sensor.getId())
                // TODO 3.使用 事件时间语义 的窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .process(
                        new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                            @Override
                            public void process(String s, Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                                long startTs = context.window().getStart();
                                long endTs = context.window().getEnd();
                                String windowStart = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss.SSS");
                                String windowEnd = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss.SSS");
                                long count = elements.spliterator().estimateSize();
                                out.collect("key=" + s + "的窗口[" + windowStart + "," + windowEnd + ")包含" + count + "条数据===>" + elements.toString());
                            }
                        }
                )
                .print();
        env.execute();
    }
}

// 自定义水位线的产生
public class CustomPeriodicWatermarkExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
                .addSource(new ClickSource())
                .assignTimestampsAndWatermarks(new CustomWatermarkStrategy())
                .print();
        env.execute();
    }
    public static class CustomWatermarkStrategy implements WatermarkStrategy<Event> {
        @Override
        public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new SerializableTimestampAssigner<Event>() {
                @Override
                public long extractTimestamp(Event element，long recordTimestamp) {
                    return element.timestamp; // 告诉程序数据源里的时间戳是哪一个字段
                }
            };
        }
        @Override
        public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new CustomBoundedOutOfOrdernessGenerator();
        }
    }
    public static class CustomBoundedOutOfOrdernessGenerator implements WatermarkGenerator<Event> {
        private Long delayTime = 5000L; // 延迟时间
        private Long maxTs = -Long.MAX_VALUE + delayTime + 1L; // 观察到的最大时间戳
        @Override
        public void onEvent(Event event，long eventTimestamp，WatermarkOutput output) {
            // 每来一条数据就调用一次
            maxTs = Math.max(event.timestamp，maxTs); // 更新最大时间戳
        }
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            // 发射水位线，默认200ms调用一次
            output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
        }
    }
}

env.getConfig().setAutoWatermarkInterval(400L);

.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(3))

.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
.allowedLateness(Time.seconds(3))
.sideOutputLateData(lateWS)

/**
 * 正常情况下，数据的产生一定是从早到晚。
 *      早产生的数据，一定是先到达系统的
 *      很少会有迟到的情况。
 *
 *      -------------------------------------
 *      水印的意义：如果当前的算子的水印是x，那么意味着正常情况下，x之前的数据都已经到达了
 *
 *      迟到：数据的时间属性<水印
 *
 *      处理迟到的数据:
 *          迟到的数据，在对应的窗口关闭之前到达，是不受影响的
 *          迟到的数据，在对应的窗口关闭之后到达，是无法进入窗口的，也无法被计算，此时可以这样处理：
 *              1、调慢水印时间
 *                  操作的是水印策略
 *                  WatermarkStrategy.forBoundedOutOfOrderNess(Duration.ofSeconds(5))
 *
 *              2、如果1无法解决，可以延迟窗口的关闭时间
 *                  操作的对象是窗口
 *                      窗口是到点就计算，然后关闭
 *                      当我延迟以后，窗口到点就计算，但是不关闭，会延迟一段时间再关闭。在此期间如果有数据，会再次触发窗口计算
 *
 *              3、如果2无法解决，可以使用测流接受迟到的数据
 *                  操作的对象是窗口
 *                  窗口关闭之后的数据，可以导入到一个测流中。后续再处理
 *
 *                  如果测流中的数据过多，说明当前的系统有问题。排查
 *                      或数据在产生时，无法保证时序，对于无法保证时序的数据，建议批处理
 *
 */
public class Demo03_HandleLate {
     public static void main(String[] args) throws Exception {
             //创建Flink配置类（空参创建的话都是默认值）
             Configuration configuration = new Configuration();
             //修改配置类中的WebUI端口号
             configuration.setInteger("rest.port",3333);
             //创建Flink环境（并且传入配置对象）
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
             //设置周期性发送水印(单位是毫秒)
             env.getConfig().setAutoWatermarkInterval(2000);
             //关闭算子链
             env.disableOperatorChaining();
             //设置并行度
             env.setParallelism(1);
             //设置水印特征
         WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
                 //设置是乱序水印
                 .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                 //从数据中抽取时间戳
                 .withTimestampAssigner(
                         (e, ts) -> e.getTs()
                 );
         //设置测流标记
         OutputTag<WaterSensor> lateTag = new OutputTag<>("late", TypeInformation.of(WaterSensor.class));
         SingleOutputStreamOperator<String> process = env.socketTextStream("hadoop102", 9999)
                 .map(new WaterSensorFunction())
                 //设置水印策略
                 .assignTimestampsAndWatermarks(watermarkStrategy)
                 //全局时间滚动窗口
                 /**
                  *  5s的滚动窗口。
                  *                  滚动窗口，从1970-1-1 0：0：0开始当作第一个窗口的起始时间计算。
                  *
                  *                  第一个窗口: [0,5000) 等价于 [0,4999]
                  *                              4999就触发第一次运算，但是不关闭。
                  *                              6999关闭
                  *                  第二个窗口: [5000,9999]
                  *                              9999就触发第一次运算，但是不关闭
                  *                              11999关闭
                  */
                 .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                 //运行延迟2s
                 .allowedLateness(Time.seconds(2))
                 //添加测流
                 .sideOutputLateData(lateTag)
                 //运算逻辑
                 .process(new ProcessAllWindowFunction<WaterSensor, String, TimeWindow>() {
                     @Override
                     public void process(ProcessAllWindowFunction<WaterSensor, String, TimeWindow>.Context context, Iterable<WaterSensor> elements, Collector<String> out) throws Exception {
                         TimeWindow window = context.window();
                         out.collect(window + ":" + MyUtils.paresToList(elements));
                     }
                 });
         //主流打印
         process.print();
         //测流打印
         process.getSideOutput(lateTag).printToErr("迟到");
         //执行
         env.execute();
     }
}