【技术分享】Apache Flink: Watermark对Window窗口的影响

点击关注“八戒技术团队”，阅读更多技术干货

Apache Flink流式数据处理中，如何针对时间维度乱序数据在Window中进行正确计算呢？准备一份含有时间维度的乱序数据。

  —-——编号3————编号2————编号1————> 接收数据顺序

Window的分配

在Apache Flink的应用程序，针对接收到的每一条数据都会计算一次窗口的归属，源码中核心算法：

假设需要一个为5s的窗口, 那么可以很简单的分配出数据的窗口归属，将size=5， globalOffset=0和staggerOffe=0,  startTime=TimeWindow.getWindowStartWithOffset(time, 0, 5), 则窗口的时间分配：[startTime, startTime+5)，参考下图：

从乱序数据中能得到一点，编号=1和编号=3的数据分配到同一个窗口window1，而编号=2的数据分配到新的window2中，并且window1早于window2生成。

由于应用程序先接收编号=1的数据，其次接收编号=2的数据，最后接收编号=3的数据。那么从数据维度来看，应用程序接收的数据是乱序的，如何保证window1触发计算的时候，window1中一定包含了编号=1和编号=3的数据呢？如果window1触发计算时，应用程序还未接收编号=3的数据时，那么计算结果永远是错误的！

Window触发计算条件

窗口触发计算的核心算法：

 从源码中可以发现，window.maxTimestamp()为窗口分配的endTime-1 <=  ctx.getCurrentWatermark() 时候，就触发了window的计算。那什么是Watermark呢？Watermark又是如何生成的呢？

 但是能得出一个结论：应用程序处理乱序数据时候，如果保证程序接收编号=3的数据后，才使 ctx.getCurrentWatermark() >= window.maxTimestamp()条件成立，那么window1触发统计的时候，就能保证编号=1和编号=3的数据一起统计出正确结果。

Watermark解释

以一个简单的示例来演示什么是Watermark以及为什么Apache Flink中要提供Watermark这种机制。

我们将以带有混乱时间戳的事件流为例，如下所示：显示的数字表达的是这些事件实际发生时间的时间戳。到达的第一个事件发生在时间t=4，随后发生的事件发生在更早的时间t=2，依此类推：

··· 23 19 22 24 21 14 17 13 12 15 9 11 7 2 4 →

假设我们要对数据流排序，我们想要达到的目的是：应用程序应该在数据流里的事件到达时就有一个算子（我们暂且称之为排序）开始处理事件。这个算子所输出的流是按照时间戳排序好的。

重新审视这些数据:

(1) 我们的排序器看到的第一个事件的时间戳t=4，但是我们不能立即将其作为已排序的流释放。因为我们并不能确定它是有序的，并且较早的事件有可能并未到达。事实上，如果站在上帝视角，我们知道，必须要等到时间戳t=2的元素到来时，排序器才可以有事件输出。需要一些缓冲，需要一些时间，但这都是值得的。

(2) 接下来的这一步，如果我们选择的是固执的一直等待，我们永远不会有结果。首先，我们看到了时间戳t=4的事件，然后看到了时间戳t=2的事件。可是，时间戳t`<2的事件接下来会不会到来呢?可能会,也可能不会。再次站在上帝视角审视数据，发现永远不会看到时间戳t` < 2的事件。最终，我们必须勇于承担责任，并发出指令，把带有时间戳t=2的事件作为已排序事件流的开始事件。

说明Apache Flink提供了一种策略：对于任何给定时间戳的事件，何时停止等待较早事件的到来，这就是Watermark: 它们定义何时停止等待较早的事件。

 Apache Flink中事件时间的处理取决于watermark生成器，后者将带有时间戳的特殊元素插入流中形成 watermarks。当事件时间t的watermark插入数据流中时，表明t之前（很可能）的事件都已经到达。

Watermark生成器

梳理Flink提供的一种默认实现源码，发现watermark的生成取决于两个核心的因素：maxTimestamp和outOfOrdernessMills参数。而maxTimestamp来源于maxTimestamp = Math.max(maxTimestamp, eventTimestamp);

那么回归到前言的乱序数据统计问题，开窗size=5s并且将outOfOrdernessMills=0时，只展示时间到秒，再来审视一下数据： 

当应用程序接收编号=1的数据时，数据编号=1分配窗口window1=[2022-02-08 15:06:50， 2022-02-08 15:06:55)，watermark=2022-02-08 15:06:52:999

当应用程序接收编号=2的数据时，数据编号=2分配窗口window2=[2022-02-08 15:06:55, 2022-02-08 15:07:00 ), watermark=2022-02-08 15:06:56:999。当watermark被编号=2的数据更新到高位，此时watermark >= window1的2022-02-08 15:06:55时，那么window1就会被立即触发计算，而不会等待后续编号=3的数据造成数据统计不准确。

因此该问题最简单的一种处理方式就只需要增加outOfOrdernessMills参数的值，让应用程序的watermark被编号=2的数据更新时，降到小于2022-02-08 15:06:55时间戳，延迟触发window1,确保window1中包含编号=1和编号=3数据。

结束语

通过上文的一个小案例，简略梳理了Apache Flink中Watermark对Window的影响。那么Watermark生成器的选择是非常重要的，究竟是使用默认实现+调参或者自定义实现接口，取决于开发者面临真实需求和具体问题。

诚邀各位IT大佬加入我们“西南名猿交流群”

一个交流技术、招聘的场地

全国程序猿皆可扫码上车噢～

原文始发于微信公众号（八戒技术团队）：【技术分享】Apache Flink: Watermark对Window窗口的影响