Flink1.11的Watermark机制

在使用eventTime的时候如何处理乱序数据&＃xff1f;我们知道&＃xff0c;流处理从事件产生&＃xff0c;到流经source&＃xff0c;再到operator&＃xff0c;中间是有一个过程和时间的。虽然大部分情况下&＃xff0c;流到operator的数据都是按照事件产生的时间顺序来的&＃xff0c;但是也不排除由于网络延迟等原因&＃xff0c;导致乱序的产生&＃xff0c;特别是使用kafka的话&＃xff0c;多个分区的数据无法保证有序。所以在进行window计算的时候&＃xff0c;我们又不能无限期的等下去&＃xff0c;必须要有个机制来保证一个特定的时间后&＃xff0c;必须触发window去进行计算了。这个特别的机制&＃xff0c;就是watermark。Watermark是用于处理乱序事件的&＃xff0c;用于衡量Event Time进展的机制。watermark可以翻译为水位线。

Watermark的核心本质可以理解成一个延迟触发机制。
在 Flink 的窗口处理过程中&＃xff0c;如果确定全部数据到达&＃xff0c;就可以对 Window 的所有数据做 窗口计算操作&＃xff08;如汇总、分组等&＃xff09;&＃xff0c;如果数据没有全部到达&＃xff0c;则继续等待该窗口中的数据全 部到达才开始处理。这种情况下就需要用到水位线&＃xff08;WaterMarks&＃xff09;机制&＃xff0c;它能够衡量数据处 理进度&＃xff08;表达数据到达的完整性&＃xff09;&＃xff0c;保证事件数据&＃xff08;全部&＃xff09;到达 Flink 系统&＃xff0c;或者在乱序及 延迟到达时&＃xff0c;也能够像预期一样计算出正确并且连续的结果。当任何 Event 进入到 Flink 系统时&＃xff0c;会根据当前最大事件时间产生 Watermarks 时间戳。

那么 Flink 是怎么计算 Watermak 的值呢&＃xff1f;

Watermark &＃61;进入Flink 的最大的事件时间(mxtEventTime)-指定的延迟时间(t)

那么有 Watermark 的 Window 是怎么触发窗口函数的呢&＃xff1f;
如果有窗口的停止时间等于或者小于 maxEventTime - t(当时的warkmark)&＃xff0c;那么这个窗口被触发执行。

其核心处理流程如下图所示。

1、本来有序的Stream中的 Watermark

如果数据元素的事件时间是有序的&＃xff0c;Watermark 时间戳会随着数据元素的事件时间按顺 序生成&＃xff0c;此时水位线的变化和事件时间保持一直&＃xff08;因为既然是有序的时间&＃xff0c;就不需要设置延迟了&＃xff0c;那么t就是 0。所以 watermark&＃61;maxtime-0 &＃61; maxtime&＃xff09;&＃xff0c;也就是理想状态下的水位 线。当 Watermark 时间大于 Windows 结束时间就会触发对 Windows 的数据计算&＃xff0c;以此类推&＃xff0c; 下一个 Window 也是一样。这种情况其实是乱序数据的一种特殊情况。

2、乱序事件中的Watermark

现实情况下数据元素往往并不是按照其产生顺序接入到 Flink 系统中进行处理&＃xff0c;而频繁 出现乱序或迟到的情况&＃xff0c;这种情况就需要使用 Watermarks 来应对。比如下图&＃xff0c;设置延迟时间t为2。

3、并行数据流中的Watermark

在多并行度的情况下&＃xff0c;Watermark 会有一个对齐机制&＃xff0c;这个对齐机制会取所有 Channel 中最小的 Watermark。

1、首先&＃xff0c;正确设置事件处理的时间语义&＃xff0c;一般都是采用Event Time。

sEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


2、其次&＃xff0c;指定生成Watermark的机制&＃xff0c;包括&＃xff1a;延时处理的时间和EventTime对应的字段。如下&＃xff1a;