【深入浅出flink】第12篇：多图全面剖析flink中的时间语义和Watermark，面试官再也难不倒我了

【深入浅出flink】第12篇：多图全面剖析flink中的时间语义和Watermark，面试官再也难不倒我了

大家好，我是雷恩Layne，这是《深入浅出flink》系列的第十二篇文章，希望能对您有所收获O(∩_∩)O

文章目录

一、Flink中的时间语义二、Watermark(水位线)

2.1 Watermark的由来2.2 什么是Watermark2.3 Watermark的生成方式2.4 Watermark在任务间的传递 三、Flink对乱序数据的三重保障

一、Flink中的时间语义

流式数据处理最大的特点是数据具有时间属性特征，Flink根据时间产生的位置不同，将时间区分为三种概念：数据生成时间（Event_time）、事件接入时间（Ingestion_time）、事件处理时间（Processing_time），用户可以根据需要选择事件类型作为流式数据的时间属性。

（1）事件时间（Event time）

事件时间是事件创建的时间，这个时间通常在到达Flink之前已经嵌入到生产数据中，因此时间顺序取决于事件产生的地方，和下游的数据处理系统的事件无关。事件时间能够保证正确性，哪怕事件是无序的、延迟的甚至是从持久层的日志或者备份中恢复的，都能保证事件得到正确处理。通常，基于事件时间的Event需要等待水位线(Watermark)的到来触发窗口。

（2）处理时间（Processing time）

处理时间是在执行算子计算过程中获取到的所在主机的时间，也就是具体执行任务的主机的系统时间。基于处理时间的流计算作业在执行时，无需等待水位线的到来触发窗口，所以可以提供较低的延迟。由于处理时间并不是数据真正产生的时间，所以在有些场景下可能会出现问题。

（3）摄入时间（Ingestion time）

摄入时间是数据接入Flink系统的时间，依赖于Source Operator所在主机的系统时钟。

处理时间（Processing time）不依赖Watermark(水位线)，Watermark只在基于事件时间和摄入时间这两种时间类型下起作用，更多的是用于事件时间。

二、Watermark(水位线) 2.1 Watermark的由来

我们知道，流处理从事件产生，到流经source，再到operator，中间是有一个过程和时间的，虽然大部分情况下，流到operator的数据都是按照事件产生的时间顺序来的，但是也不排除由于网络、分布式等原因，导致乱序的产生，所谓乱序，就是指Flink接收到的事件的先后顺序不是严格按照事件的Event Time顺序排列的。

那么此时出现一个问题，一旦出现乱序，如果只根据eventTime决定window的运行，我们不能明确数据是否全部到位，但又不能无限期的等下去，此时必须要有个机制来保证一个特定的时间后，必须触发window去进行计算了，这个特别的机制，就是Watermark。

2.2 什么是Watermark

Watermark被称为水位线，是一种衡量事件时间(Event time)进度的机制，本质上是一种时间戳，可以在读取 Source时候指定或者在transformation *** 作之前，用Watermark生成器按照需求指定。

具体来说，Watermark有如下特点：

Watermark是一种衡量事件时间(Event time)进度的机制，本质上是一种时间戳。Watermark是用于处理乱序事件的，而正确的处理乱序事件，通常用Watermark机制结合window来实现。也就是当最大的事件时间maxEventTime小于Watermark之前，window仍然可以接收数据，根据Event time将其分配到指定的窗口。Watermark也可以理解成一个延迟触发机制，每个Watermark都可以设置一个延时时长，表示事件延迟触发的程度。假设Watermark的延时时长t，每次系统会校验已经到达的数据中最大的maxEventTime，然后认定eventTime小于maxEventTime - t的所有数据都已经到达，如果有窗口的停止时间等于maxEventTime – t，那么这个窗口被触发执行。

窗口实际上是由其Trigger触发的，EventTimeTrigger中定义了maxEventTime达到或超过Watermark时触发，具体实现可参考我的博客：flink窗口计算的触发器

当Flink接收到数据时，会按照一定的规则去生成Watermark，这条Watermark就等于当前所有到达数据中的maxEventTime - 延迟时长（只有maxEventTime更新后，才会生成WaterMark）。也就是说，Watermark是基于数据携带的时间戳生成的，一旦Watermark比当前未触发的窗口的停止时间要晚，那么就会触发相应窗口的执行。由于event time是由数据携带的，因此，如果运行过程中无法获取新的数据，那么没有被触发的窗口将永远都不被触发。

下面我们用一组图来说明Watermark的生成过程，设置桶大小(即窗口大小)为5s，第一个窗口范围为[0,5)，延迟时长t为3s，然后每次收到一个Event，计算当前的Watermark：

（1）收到第1个Event，提取其EventTime为1，maxEventTime=1，Watermark=maxEventTime-t=1-3=-2

（2）收到第2个Event，提取其EventTime为4，maxEventTime=4，Watermark=4-3=1

（3）收到第3个Event，提取其EventTime为3，maxEventTime不更新，仍为4，此时不生成Watermark。

（4）收到第4个Event，提取其EventTime为5，maxEventTime=5，Watermark=5-3=2

（5）收到第5个Event，提取其EventTime为2，maxEventTime不更新，仍为5，此时不生成Watermark。

（6）收到第6个Event，提取其EventTime为6，maxEventTime=6，Watermark=6-3=3

（7）收到第7个Event，提取其EventTime为8，maxEventTime=8，Watermark=8-3=5，此时第一个窗口范围为[0,5)，已经小于Watermark，触发其窗口计算、得到结果、关闭窗口。

（8）收到第8个Event，提取其EventTime为7，maxEventTime不更新，仍为5，此时就剩一个窗口了

需要注意的是，如果我们在算子上设置allowedLateness（允许处理迟到的数据），假如设置参数为1min，那么运行1min处理迟到的数据，所以在时间戳为8的event到来后，[0,5)的bucket只会输出结果，并不会关闭窗口，只有时间戳为68的数据来了之后（此时waterMark=65）才会关闭[0,5)的bucket。

2.3 Watermark的生成方式

Flink中生成水位线的方式有两种：Periodic Watermarks（周期性）和Punctuated Watermarks，上图举例用的是Punctuated Watermarks式的生成方式。

Punctuated Watermarks：来一个数据生成当前的Watermark，flink没有提供它的实现类，我们可以自定义实现Periodic Watermarks：系统周期性生成并返回当前的Watermark，它返回的Watermark仅在大于上一次返回的Watermark情况下有效。

后面我将单独写一篇博客详细介绍这两种生成方式的源码实现。

2.4 Watermark在任务间的传递

由于每一个算子的SubTask执行进度不同，它们的Watermark也不同，一个SubTask的Watermark更新后，才向它所有的下游任务broadcast(广播)，发送自己的Watermark，具体流程如下图所示：

每一个Task接收其所有上游任务发来的Watermark，选出其中最小Watermark当做自己的Watermark。只有当前Task的Watermark更新了，才会向下游广播。

flink之所以采用这样的Watermark传递机制，即拿到上游任务最小的Watermark，是为了得知当前EventTime的最低进度，这样才能不遗漏上游任何任务发来的数据。

三、Flink对乱序数据的三重保障

我们思考一个问题：怎样避免乱序数据带来计算不正确性？

常用的解决办法是：当最大的事件时间maxEventTime达到了窗口关闭时间，不应该立刻触发窗口计算，而是等待一段时间，等迟到的数据来了再关闭窗口。

但是，我们应该等待多久的时间呢？由于网络、分布式等原因造成的延时，一般大多数迟到的数据都会在最近一段时间到来，这个最近一段时间一般是毫秒级的，Watermark就是做到了这样的保障。还有很少的一部分数据会迟到很久，我们可以通过allowedLateness和sideOutputLateData来兜底。

处理乱序数据，三重保证机制：

（1）Watermark

能够保证迟到很短的时间的数据到来后（一般是迟到毫秒级别内的数据，最大不超过1s），触发窗口关闭并输出。（即能够hold住短时间内迟到的数据）

（2）allowedLateness

如果设置allowedLateness，比如设置2min，这样Watermark时间到之后会先输出一个近似的结果，然后在2min内来一个数据更新一次结果，等待时间2min到了之后，关闭窗口。

（3）sideOutputLateData
再之后迟到的数据，可以通过sideOutputLateData（侧输出流）来兜底。

一般来说，Watermark 用来让程序自己平衡延迟和结果正确性。Watermark的值不能太大也不能太小，比如将Watermark设置2min，这样相当于代替了allowedLateness的功能，并且这样就相当于一直等到迟到2min的数据过来后，窗口才能输出，中间不会输出结果，相当于大大增加了统计的延迟。如果将Watermark设置的很小，比如5ms，这样可能很多迟到的数据会在allowedLateness或sideOutputLateData下完成，而allowedLateness或sideOutputLateData是来一个数据更新并输出一个结果，会大大增加计算的成本。

allowedLateness或sideOutputLateData机制下，来一个数据数据会触发窗口计算，并打印结果，而Watermark是窗口结束时间小于Watermark 才触发计算，输出结果。