【入门Flink】- 09Flink水位线Watermark

在窗口的处理过程中，基于数据的时间戳，自定义一个“逻辑时钟”。这个时钟的时间不会自动流逝；它的时间进展，就是靠着新到数据的时间戳来推动的。

什么是水位线

用来衡量事件时间进展的标记，就被称作“水位线”（Watermark）。

具体实现上，水位线可以看作一条特殊的数据记录，它是插入到数据流中的一个标记点，主要内容就是一个时间戳，用来指示当前的事件时间。而它插入流中的位置，就应该是在某个数据到来之后；这样就可以从这个数据中提取时间戳，作为当前水位线的时间戳了。

有序流中水位线

（1）理想状态（数据量小），数据应该按照生成的先后顺序进入流中，每条数据产生一个水位线；

（2）实际应用中，如果当前数据量非常大，且同时涌来的数据时间差会非常小（比如几毫秒），往
往对处理计算也没什么影响。所以为了提高效率，会每隔一段时间生成一个水位线。

乱序流中水位线

在分布式系统中，数据在节点间传输，会因为网络传输延迟的不确定性，导致顺序发生改变，这就是
所谓的“乱序数据”。

（1）乱序+数据量小：还是靠数据来驱动，每来一个数据就提取它的时间戳、插入一个水位线。乱序数据，插入新的水位线时，要先判断一下时间戳是否比之前的大，否则就不再生成新的水位线。也就是说，只有数据的时间戳比当前时钟大，才能推动时钟前进，这时才插入水位线。

（2）乱序+数据量大：考虑到大量数据同时到来的处理效率，可以周期性地生成水位线。这时
只需要保存一下之前所有数据中的最大时间截，需要插入水位线时，就直接以它作为时间戳生成新的水位线。

（3）乱序+迟到数据：无法正确处理“迟到”的数据。为了让窗口能够正确收集到迟到的数据，设置迟到时间，比如2秒：也就是用当前已有数据的最大时间戳减去2秒，就是要插入的水位线的时间戳。9秒的数据到来之后，事件时钟不会直接推进到9秒，而是进展到了7秒；必须等到11秒的数据到来之后，事件时钟才会进展到9秒，此时迟到2秒的数据也会被正确收集处理。【迟到时间不能设置过长，否则会对实时性会有所影响】

水位线的特性

水位线是插入到数据流中的一个标记，可以认为是一个特殊的数据
水位线主要的内容是一个时间戳，用来表示当前事件时间的进展
水位线是基于数据的时间戳生成的
水位线的时间戳必须单调递增，以确保任务的事件时间时钟一直向前推进
水位线可以通过设置延迟，来保证正确处理乱序数据
一个水位线Watermark(t)，表示在当前流中事件时间已经达到了时间戳t，代表t之前的所有数据都到齐了，之后流中不会出现时间t'≤t的数据

水位线与窗口配合，完成对乱序数据的正确处理。

水位线是流处理中对低延迟和结果正确性的一个权衡机制。

水位线生成策略

生成水位线的方法：.assignTimestampsAndWatermarks()，主要用来为流中的数据分配时间戳，并生成水位线来指示事件时间。【指定水位线生成策略】

stream.assignTimestampsAndWatermarks(<watermark strategy>);

WatermarkStrategy 水位线策略是一个接口，里面内置一些生成策略：

有序流中内置水位线设置

时间戳单调增长，所以永远不会出现迟到数据的问题。WatermarkStrategy.forMonotonousTimestamps()

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy.
                <WaterSensor>forMonotonousTimestamps()
                // 指定时间戳分配器，从数据中提取 单位毫秒
                .withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (element, recordTimestamp) -> {
                    System.out.println(" 数据 =" + element + ",recordTs=" + recordTimestamp);
                    return element.getTs() * 1000L;
                });

乱序流中内置水位线设置

由于乱序流中需要等待迟到数据到齐，必须设置一个固定量的延迟时间。WatermarkStrategy. forBoundedOutOfOrderness()

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
                // 乱序数据，等待3s
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                // 指定时间戳分配器，从数据中提取 单位毫秒
                .withTimestampAssigner((SerializableTimestampAssigner<WaterSensor>) (element, recordTimestamp) -> {
                    System.out.println(" 数据 =" + element + ",recordTs=" + recordTimestamp);
                    return element.getTs() * 1000L;
                });

自定义水位线生成器

（1）周期性水位线生成器（Periodic Generator）

周期性生成器一般是通过 onEvent()观察判断输入的事件，而在onPeriodicEmit()里发出水位线。
模仿该类BoundedOutOfOrdernessWatermarks

public class CustomBoundedOutOfOrdernessGenerator<T> implements WatermarkGenerator<T> {

    private Long delayTime = 5000L; // 延迟时间
    private Long maxTs = Long.MIN_VALUE + delayTime + 1L; // 观察到的最大时间戳

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        // 每来一条数据就调用一次
        maxTs = Math.max(eventTimestamp, maxTs); // 更新最大时间戳
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // 发射水位线，默认 200ms 调用一次
        output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
    }
}

在 onPeriodicEmit()里调用 output.emitWatermark()，就可以发出水位线了；方法由系统框架周期性地调用，默认 200ms 一次。【不建议修改】

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

（2）断点式水位线生成器（Punctuated Generator）

断点式生成器会不停地检测 onEvent()中的事件，当发现带有水位线信息的事件时，就立即发出水位线。

如下：只要有数据来就直接发射水位线

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        // 每来一条数据就调用一次
        maxTs = Math.max(eventTimestamp, maxTs); // 更新最大时间戳
        output.emitWatermark(new Watermark(maxTs - delayTime - 1L));
    }

（3）在数据源中发送水位线

可以在自定义的数据源中抽取事件时间，然后发送水位线。

自定义数据源中生成水位线和在程序中使用assignTimestampsAndWatermarks 方法生成水位线二者只能取其一。

env.fromSource(
kafkaSource,
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)), 
// WatermarkStrategy.noWatermarks() 或者不发送水位线
"kafkasource"
)

水位线的传递（空闲等待withIdleness）

一个任务接收到多个上游并行任务传递来的水位线时，应该以最小的作为当前任务的事件时钟。

如下案例：当程序并行度设置为2时，自定义分区器导致一个分区一直拿不到数据（最小时钟一直为null），此时如不加以干预，事件时钟将永远不会推进，存在问题。设置空闲时间，当超过空闲时间一直收不到该分区数据，直接忽略该分区，还是会依旧推进时间时钟。

        env.setParallelism(2);
		// 自定义分区器：数据%分区数，只输入奇数，都只会去往map 的一个子任务
        SingleOutputStreamOperator<Integer> socketDS = env.socketTextStream("xxxx", 7777)
                .partitionCustom(new MyPartitioner(), r -> r).map(Integer::parseInt)
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Integer>forMonotonousTimestamps().withTimestampAssigner((r, ts) -> r * 1000L)
                                .withIdleness(Duration.ofSeconds(5)) // 空闲等待 5s
                );

	    // 分成两组：奇数一组，偶数一组，开 10s 的事件时间滚动窗口socketDS
        .keyBy(r -> r % 2)
        .window(TumblingEventTimeWindows.of(Time.seconds(10)))
        ...

迟到数据的处理

1）推迟水印推进（设置延迟时间）

水印产生时，设置一个乱序容忍度，推迟系统时间的推进，保证窗口计算被延迟执行，为乱序的数据争取更多的时间进入窗口。

WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10));

2）设置窗口延迟关闭

Flink 的窗口，也允许迟到数据。当触发了窗口计算后，会先计算当前的结果，但是此时并不会关闭窗口。当达到设置延迟关闭时间之后，才会真正关闭窗口，关闭窗口后再迟到的数据就不会再处理。

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

3）使用侧流接收迟到的数据

最后兜底，窗口关闭之后的迟到数据，使用侧输出流输出。

完整方案：

public class WatermarkLateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("124.222.253.33", 7777)
                .map(new WaterSensorMapFunction());
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
          		 // 1.设置迟到时间 3s
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner((element, recordTimestamp) -> element.getTs() * 1000L)
                // .withIdleness(Duration.ofSeconds(5)); // 空闲等待 5s;

        SingleOutputStreamOperator<WaterSensor>
                sensorDSWithWatermark = sensorDS.assignTimestampsAndWatermarks(watermarkStrategy);
        OutputTag<WaterSensor> lateTag = new OutputTag<>("latedata", Types.POJO(WaterSensor.class));
        SingleOutputStreamOperator<String> process = sensorDSWithWatermark.keyBy(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10))) 
                .allowedLateness(Time.seconds(2)) // 2.推迟2s关窗
                .sideOutputLateData(lateTag) // 3.关窗后的迟到数据，放入侧输出流
                .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);
                            }
                        }
                );
        process.print();
        // 从主流获取侧输出流，打印
        process.getSideOutput(lateTag).printToErr("关窗后的迟到数据");
        env.execute();
    }
}