流式计算-low watermark机制

前言

最近在学习流式计算相关的知识，在阅读了MillWheel论文，在这里对low watermark机制做一个总结与介绍。

Window

实时数据流是永不停歇的，我们无法获取所有的数据并产出一个最终的结果。很多情况下我们更关心的是最近的状态，而不是从实时数据流启动至今的统计数据。因此需要将数据流切分为一个个片段，这些片段形象的称为Window。总而言之，Window是将无边界的实时数据流进行划分的一种机制。

Window大体分为三种：

Time Window：依靠时间作为划分Window的依据，意思就是设立一个时间间隔来结束window，进行计算
- Fixed Window：固定窗口，数据彼此不重叠
- Sliding Window：滑动窗口，数据会重叠

Session Window：依靠用户的会话作为划分Window的依据
session window主要依靠设立的时间间隔来划分一个window，即session gap。session window没有具体的结束和开始时间，当一个window中超过session gap时间还没有元素到来，该window就会结束并进行计算，此时就称为一个会话。

Global Window：将具有相同key的数据分配给同一个窗口，需要用户自定义触发器来进行计算，否则该window不会结束。

从中可以看出，Window的计算是非常依赖Time的。对于一个流式计算系统来说，Time分为两类，选择哪种作为划分Window的标准更好，接下里我们就介绍两种不同的Time。

Processing/Event Time

Processing Time：数据被处理的时间
Event Time：数据产生的时间
这两个概念很好理解，Processing Time就是指数据在流式计算中，不同PE对其进行处理的时间。而Event Time就是指数据产生的时间，比如用户点击url的时间。

在流式计算早期，一般采用Processing Time作为Window划分的时间，但是这样有一个问题，当数据在上游延迟时，以Processing Time作为时间进行划分Window，就会出现误差。比如下图中的例子：

本来数据C在t3时刻产生，应当划分进t1-t3这个Window，但是由于某些原因（网络延迟等），导致数据C在t7时刻才被处理。如果按照Processing Time进行处理，那么C就落在了t4-tn的Window中，这很明显会导致计算结果的不准确。比如我们需要统计在某个时间段用户点击广告的次数，如果按照Processing Time来算，那可能有些数据就会被划分到其他window中，因此需要记录Event Time，做法也很简单，让数据携带timestamp就行了。

使用了Event Time虽然可以让数据落在正确的Window中，但是由于网络存在延迟以及各种原因，数据仍可能会迟到，如何才能让Window尽可能等待这些会迟到的数据，这也是需要考虑的问题。最直观的做法就是设置一个等待时间，但等待时间设置成多少合理，也需要考虑。为了解决这个问题，MillWheel论文中提出了lower watermark机制。

Timer

Timer即定时器，作为Window的触发源，告知Window应当开始计算了。Timer分为两类：

WallTime Timer：即按照正常的现实时间作为触发源
LowWatermark Timer：以低水位作为触发源

low watermark

low watermark其实就是一个时间戳，每个计算节点都会维护这样一个时间戳作为low watermark。为了解决这类问题，MillWheel采用了Low Watermark机制，将lwm timer作为Window的触发源。Low Watermark机制是流式系统中解决数据的完整性以及时效性问题的一种较好的方案，即为每个计算组件设立低水位值（时间戳），确保了不存在比当前时间戳还晚到达的数据，该机制的确保跟它的计算公式有关。

假设有计算节点A和C，并且C是A的上游节点，则A的低水位值的计算应该遵从以下公式。

low watermark of A = min (oldest work of A , low watermark of C : C outputs to A )

从中可以看出，A的低水位值不只和A本身的最旧数据有关，也跟上游的低水位有关。因此，只要上游还有更旧的数据存在，就会通过低水位机制维护的low watermark告知下游，下游便会更新它自己的low watermark并且由于lwm timer未触发，因此会进行等待。这样就明显形成了一个递归结构，low watermark的值与数据流的Injector有一定的关联性。

该机制是怎样解决这个问题的，这里还是给出一个例子帮助大家理解。

如图所示，A的上游有C1-Cn各点，一直追溯到源头Ij1-Ijn。A中维护了一个Window，用于统计9:30-10:00这半个小时的数据，而lwm timer为10:00，由于此时A的lwm为9:50，还没有到lwm timer，因此Window不会关闭，会等待上游滞留的数据到达。lwm(A)之所以为9:50的原因是上游的数据有延迟，min(lwm of C1…Cn)=9:50。如果按照墙上时间10:30，此时早就应该触发了，便会导致结果的不准确。

总结

关于流式计算的low watermark机制大概就是这样，它可以在一定程度上保证数据的完整性以及时效性。但实际上，若就是有数据比low watermark还晚到达仍没办法解决，比如数据在没有进入到流式计算系统之前就延误了，那low watermark根本不得而知。Flink为了尽可能解决这种情况，除了low watermark还设置了allow lateness参数，即Window被lwm timer触发后，还会等待allow lateness时间才开始计算，但这样很明显会损失一定的实时性。

由于每个流式计算系统的实现方式不一样，MillWheel又将low watermark分为两种watermark并且采用Server/Client中心化管理lwm，这里就不展开了。