谈谈EventTime以及Watermark

多次被网友问及EventTime（事件时间）以及Watermark（水位线）的问题。今天开篇文章来谈一谈个人观点，如果有表述不当的地方，希望各位指正。

什么是EventTime

简单来说就是：事件发生于产生设备上的本地时间。作为对比，需要再引出另一个概念：ProcessingTime，它指的是事件被处理时所在处理节点的本地时间。

关于EventTime和ProcessingTime，有个很好的类比比较贴切。那就是经典的《星球大战》系列电影的发布顺序跟观影顺序的问题。

这个不错的类比思路来自于Flink社区的Kostas Kloudas。

《星球大战》系列电影发布时间点可以看作是EventTime：
《星球大战4·新希望》(1977)
《星球大战5·帝国反击战》(1980)
《星球大战6·绝地大反攻》(1983)
《星战前传1·幽灵的威胁》(1999)
《星战前传2·克隆人进攻》(2002)
《星战前传3·西斯的复仇》(2005)
《星球大战7》(2015年)
另外，预计：
《星球大战8》(2017年)
《星球大战9》(2019年)

但是对于一个不是很熟悉连贯剧情的人而言，他可能会按照1,2,3,4,5…的自然顺序观看（这里只是举个例子进行假设，其实关于观影顺序，网上也有各种讨论）。假设每一部都是一个事件，你观看一部类似于处理一个事件，那么你的ProcessingTime跟它原先的EventTime是不一致的。

为什么需要EventTime

假设你有个需求：需要计算一个网站的QPS然后绘制出变化曲线图，假设访问请求被记录到日志文件，然后最终通过流处理系统来计算、统计。因为某些原因，你的流处理系统出现故障，导致其将会下线一段时间，比如10分钟。那么在下线的这段时间内的事件可能堆积在持久存储里，比如文件系统或者消息系统（假如采集agent仍然继续工作）。当你的流处理系统恢复并重新上线后。如果你不以日志的EventTime作为时间基准，而是基于ProcessingTime，那么这中断的10分钟的请求日志就仿佛是突然到来的请求一样。因此，绘制的曲线图将会呈现一个非常短区间的尖锐的脉冲，而中断的那段时间则几乎是零，但这个图表显然是不符合事实的。这里时间戳的不一致就是由于选择的时间参照体系的不同。

上面的这个场景其实是流处理系统处理历史数据的场景，在实时事件流场景中，理论上EventTime跟ProcessingTime应该是一致的（也就是发生时间即处理时间），但现实也不是一致的，原因也有很多（比如，资源竞争，网络拥塞等等），这两个时间的偏差被称之为时间歪斜（skew）。

关于乱序

事件流可能是乱序（out-of-order）的，这跟EventTime和ProcessingTime无关，这是客观事实，关于乱序，不同的人有不同的理解。常容易因为理解不一致而导致沟通偏差。因为看起来它本身就有点模糊，或者说产生事件乱序的情况较多。

就一个独立的事件源设备而言（为了方便讨论，从流处理的角度来看我们将其称之为一个物理流），它本身并不能保证是有序的。这里涉及到你发送事件的机制（单线程？同步发送？Ack、网络）等。

而有些人认为的乱序是，假设上面这种情况有序的前提下，事件可能有多个源（多个生产者处于不同的设备），但它们仍可被视为一个逻辑流，那么它们也有可能因为网络延迟或者抖动，或者机器失败等造成延迟或者失序。

还有人谈及的无序会涉及到具体的技术细节里，比如类似于Kafka里的topic、partition如何设计&存储数据，甚至流的source并行等…

每一种都可能是某个人所理解的乱序，所以更高效的沟通方式是，确保沟通双方就乱序这一看法达成一致后再讨论。

其实，大部分你看到的跟EventTime一起出现的“乱序”这个词，通常所描述的乱序的原因是：“网络”或者“机器”出现异常的情况，也就是说如果没有这些状况，那么它应该是“有序”的。所以，你也就应该知道以上哪些并不是EventTime所讨论的上下文中的“乱序”了。

什么是Watermark

现在我们知道了为什么要引入EventTime，不同的时间语义的选择归根到底其实是参照系的选择。

接下来我们看有了EventTime为何要有Watermark呢？因为流是无界数据集，而我们通常需要进行一些聚合运算，这两者是冲突的，因为无论是从时间还是空间上，都不现实，所以计算还必须回归到有界数据集上，因此就引入了窗口来作为框定一个数据集的界限。基本上窗口都可看作是时间窗口（计数窗口某种意义上也可看作是逻辑时间窗口的特例），具有时间属性的窗口，肯定要基于时间来触发计算了。但是如果你不以ProcessingTime作为参照，流处理系统怎么知道何时触发窗口呢？因此，为了告知窗口应该什么时候触发，那么就需要在EventTime中找到一个基准，该基准用来界定这样一个事实：从当前时间点开始，后来的事件没有比当前事件的EventTime更小的了，它就称之为“Watermark”。当然既然是作为一个界定的基准，那么它必然是不同于数据流中的数据的，在Flink中它被定义为一种特殊的“事件”混杂在数据流中流向下游。窗口就以此作为触发的依据。但是，需要注意EventTime和Watermark其适用范围是整个Streaming领域的。

Watermark与乱序

Watermark有助于解决乱序问题。Watermark作为EventTime进度的一种基准（信号），它描述的是“当前最小”的问题，但是它并不规定Watermark本身对应的这个事件之前的事件的顺序，它们可以是有序的，也可以是无序的。因此，对于有序流和无序流它都能起到界定作用。所以把时间线拉长，然后提取出由Watermark所组成的Watermark流（也可以将Watermark所界定的时间段内的事件都归拢于最近的Watermark），肯定是有序的。

Watermark能完全解决乱序问题吗？有了Watermark，基于EventTime的窗口得以很好地运转。但毫无疑问，Watermark的界定逻辑太理想化了，凡事总有例外，你永远不知道一个乱序数据流会乱序到什么程度，也就是某个小于Watermark的事件要延迟多久才到达。也许你的窗口已经触发计算完成了一会儿，它来了，也许很久之后它才到来。那么这些延迟事件，你是支持对它们的处理呢还是不支持？我就可以明确地告诉你这一点，Flink是支持的，迟到事件也是事件，我们怎么能不支持呢？它允许你定义一个最大的延迟时间。也就是在它最大允许的延迟时间内到达的迟到事件，它还是认的。那么迟到更久的呢？呵呵，没办法，这肯定是权衡的结果，你不可能支持无限制迟到的事件。在允许延迟的范围内到来的迟到事件会导致窗口可能重新被计算。这里为了结果的正确性，按照Dataflow的模型，窗口应该是支持“累积和撤回”的，但目前从Beam给出的结果来看，都没有得到很好的支持。

最小Watermark的问题

如果你在生产端有多个事件生产者分布在不同的设备上，那么毫无疑问，每个设备上事件的Watermark也是不同的，那么为了保证消费者处理的正确性你需要对来自每个设备上事件流的Watermark进行跟踪，这有时对流处理器而言将会变得很棘手，尤其是生产者设备的新增和移除的情况下。

无论是一个物理流还是一个逻辑流，为了利用分布式的计算能力，它们将会被shuffle，然后在不同的节点上并行计算，同时也会使Watermark的跟踪变得复杂。针对多种Watermark的场景，常用的做法是选择最小Watermark来定义当前的逻辑Watermark，按照Watermark本身的语义也是合理的。这里的逻辑Watermark可能是跨节点全局意义的Watermark，也可能是单节点自身的Watermark。Flink对来自多个input channel的Watermark就选择最小的Watermark作为节点的Watermark基准，但这种情况下，多个并行的同质的任务实例，其实还是存在各自不同的Watermark，它们之间不会同步。与这个设计截然不同的是，跟Flink同处一个领域的另外框架Gearpump选择了跨节点的“全局最小逻辑时钟”的设计。

这一不同的设计抉择也对其他设计产生了影响，比如检查点。Gearpump采用的这种设计大大简化了检查点保证“exectly-once”的设计，它基于全局Watermark来触发检查点的执行。而Flink的ABS算法基于周期性地触发+Barrier（路障）+align（对齐），它必须要以barrier+align将流隔离成多个stage，单从这点来看这种机制明显要复杂得多。但是综合来看，Flink的这种做法更灵活。首先，它本身支持三种时间语义（它必须保证检查点机制适配这三种语义），另外Barrier的align或者track，提供了不同级别的delivery一致性保证。而且，周期性的这种做法，会使得检查点的执行时间以及状态所需的存储空间，这些变动都是可预期的。而这些对Gearpump而言，会显得有些不足。

EventTime的准确度与可靠性

EventTime一定是准确、可靠的么？答案当然是否定的。比如，就将EventTime的设备假想成你的iPhone手机，不管是否有意它是可更改而且会偏离标准时间的（假设以网络标准时间为基准）。那么毫无疑问，它也不是绝对可靠。

那么此时，你怎么来矫正设备不正确的时钟呢？这时，假设你流处理系统中接收事件的节点本地时钟是可靠的（它跟网络标准时间做了同步），可以记住三个特定的时间戳：

事件在设备上的发生时间戳，标记为t1；
事件从设备上被发送出去的时间戳，标记为t2；
事件被流处理系统接收到的时间戳，标记为t3；

这里假设网络传输的时间极短，可忽略不计。

那么，你用t3减去t2的偏差就可以近似看作事件设备跟基准时间的偏差。然后，再将这个偏差应用到t1上，就可以用来纠正出正确的事件时间[1]。

[1] Alex Dean: “Improving Snowplow’s Understanding of Time,” snowplowanalyt‐ics.com, September 15, 2015.

以上，谈了一些观点、看法，也算部分答网友问。如果有不同的观点或者表述不正确的地方，也欢迎指正&讨论。谢谢~

微信扫码关注公众号：Apache_Flink

QQ扫码关注QQ群：Apache Flink学习交流群（123414680）