Flink：watermark

Table of Contents

三种时间概念

Processing time

Event Time

Ingestion time

watermark

并行流的Watermarks

迟到的事件

watermark分配器

watermark的两种分配器

三种时间概念

在谈watermark之前，首先需要了解flink的三种时间概念。在flink中，有三种时间戳概念：Event Time 、Processing Time 和 Ingestion Time。其中watermark只对Event Time类型的时间戳有用。这三种时间概念分别表示：

Processing time

处理时间，指执行算子操作的机器的当前时间。当基于处理时间运行时，所有关于时间的操作（如时间窗口）都将使用执行算子操作的机器的本地时间。例如，当时间窗口为一小时时，如果应用程序在9:15 am开始运行，则第一个窗口将包括在9:15 am到10:00 am之间被处理的事件，下一个窗口将包含在10:00 am到11:00 am之间被处理的事件，依此类推。

处理时间是最简单的时间概念，不需要流和机器之间的协调。它提供了最佳的性能和最低的延迟。但是，在分布式和异步环境中，处理时间不能提供确定性，因为它容易受到上流系统（例如从消息队列）到达Flink的速度、flink内部operators之间交互的速度，以及中断（调度或其他情况）等因素的影响。

Event Time

事件时间，是每个event在其生产设备上产生的时间，即元素在到达flink之前，本身就自带的时间戳。

所以说，Event Time的时间戳取决于数据，而与其他时间无关。使用Event Time，必须在从执行环境中先引入EventTime的时间属性。如：

val env = StreamExecutionEnvironment.getExecutionEnvironment
// 从调用时刻开始给env创建的每一个stream追加时间特征
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

然后通过Dastream的assignTimestampsAndWatermarks方法指定event time时间戳，具体操作不做赘述。

在理想情况下，事件时间是有序的。但实际上，由于分布式操作，以及网络延迟等原因，event可能不是按照event time的顺序到达的。所以flink对处理乱序数据的方案是提供一个允许延迟时间，在允许延迟时间内到达的元素将会重新触发一次计算。这个延迟时间时相对event time而不是其他时间的，而event time不是由flink决定的，那么如何判断当前的event time到底时多少呢？flink通过一个watermark来确定与维护当前event time的最大值。这也是本文将会在后面重点解释的。

Ingestion time

Ingestion time是event进入Flink的时间，即执行source操作时的时间。

Ingestion time从概念上讲介于Event Time和Processing time之间。

与Processing time相比，它花费的资源会稍微多一些，但结果却更可预测。由于 Ingestion time使用稳定的时间戳（仅在addSource处分配了一次），因此对记录的不同窗口操作将引用相同的时间戳，而在Processing time中，每个窗口操作都会更新事件的Processing time，所以可能一个上游窗口中的记录会分配给不同的下游窗口（基于本地系统时钟和任何可能的延误）。

与Event Time相比，Ingestion time程序无法处理任何乱序事件或迟到的数据，但是程序不必指定如何生成watermarks。

下图为三种时间语义的图解：

watermark

上面说到，支持Event Time需要一种测量时间进度的方法，用于判断当前event time的时间。这个机制就是watermark。

watermark会根据数据流中event的时间戳发生变化。watermark意味着当前流中已经到达的event的最大时间戳，也就是说，往后到达的event的时间戳应该要大于watermark，或者说时间戳小于watermark的数据都已经到达了。

如下图例子中，事件是按顺序排列的（相对于其时间戳），理想情况下，watermark周期性维护着当前event的最大时间戳：

但是，通常情况下，event都是乱序的，不按时间排序的，~~通常，watermark用于声明某个时间点，表示某个时间点前的数据都应该到达了~~（官网上是这么说的，官网描述的有点模糊，实际上对于乱序事件，一般会结合允许延迟机制，触发计算的条件是：watermark = 窗口的endtime + 最大允许延迟时间。所以官网上实际上是说watermark表示某个时间点是指endtime+允许延迟，则endtime前的元素都应该到达），一旦watermark到达触发计算的时间点，那么窗口就会把已经到达的event中，时间戳小于endtime的event进行计算。如下图所示：

那么，watermark到底是以怎样的一种形式存在的呢？实际上，watermark就是一种特殊的event，它被参杂在Dstream中，watermark由flink的某个操作生成后，就在整个程序中随event一同流转，如下图所示：

以下是watermark的代码，可以看出watermark的就是一个流元素，仅包含一个时间戳属性：

public final class Watermark extends StreamElement {/** The watermark that signifies end-of-event-time. */public static final Watermark MAX_WATERMARK = new Watermark(Long.MAX_VALUE);// ------------------------------------------------------------------------/** The timestamp of the watermark in milliseconds. */private final long timestamp;/*** Creates a new watermark with the given timestamp in milliseconds.*/public Watermark(long timestamp) {this.timestamp = timestamp;}

并行流的Watermarks

watermark可以在source处生成（也可以在source后通过其他算子生成，如map、filter等），如果source有多个并行度，那么每个并行度会单独生成一个watermark，这些watermark定义了各分区的event time。
当并行度发生变化（即上游的一个分区可能被下游多个分区使用时），每个分区的watermark是会广播至下游的每个分区的，如一些聚合多个流的操作，如 keyBy(…) 或者partition(…)，此类操作的watermark是在所有输入流中取最小的watermark。当带有watermark的流通过此类算子时，会根据每个分区的watermark来更新watermark。

举个例子：当上游并行度数为4，下游的某个分区的窗口中的watermark如下：

1.当已到达的watermark分别为2、4、3、6时，窗口中的watermark为2，触发watermark为2的对应窗口计算，并将watermark=2广播至下游。

2.当第一个窗口的watermark被更新为4时，所有分区中已到达最小的watermark是3，则将窗口的watermark更新为3，触发对应窗口的计算，并将watermark=3广播至下游。

3.当第二个分区的watermark被更新为7，所有分区中已到达最小的watermark还是3，不做处理。

4.当第三个分区的watermark被更新为6，所有分区中已到达最小的watermark是4，则将窗口的watermark更新为4，触发对应窗口的计算，并将watermark=4广播至下游。

下图显示了event和watermark在一个并行流的示例，以及算子如何跟踪事件时间的：

迟到的事件

在介绍watermark时，提到了现实中往往处理的是乱序事件，即当event处于某些原因而延后到达时，会发生该event time<watermark的情况，显然，这是有违watermark的制定条件的，所以flink对处理乱序事件的watermark有一个允许延迟的机制，允许在一定事件内迟到的时间仍然视为有效事件。

watermark分配器

当watermark完全基于event time时，如果没有元素到达，则watermark不会被更新，这就说明，当一段时间没有元素到达，则在这个时间间隙内，watermark不会增加，那么也不会触发窗口计算。显然，如果这段时间很长的话，那么该窗口中已经到达的元素将会等待很久才会被输出计算结果。

为了避免这种情况，可以使用周期性的watermark分配器（AssignerWithPeriodicWatermarks 下面马上提到），这些分配器不仅仅基于event time进行分配。比如，可以使用一个分配器，当一段时间没有接收到新的event时，则将当前时间作为watermark。

watermark的两种分配器

flink生成watermark有两种机制：

AssignerWithPeriodicWatermarks ：分配时间戳并定期生成watermark（可以取决于event time，或基于处理时间）。
AssignerWithPunctuatedWatermarks：分配时间戳并根据每一个元素生成watermark（每来一个元素都进行一次判断，相更消耗性能）

通常情况下会使用第一种机制，原因除了更节省性能外，在上面的分配器中也有提到。

下面分别对两种机制进行介绍。

AssignerWithPeriodicWatermarks

对每个元素都调用extractTimestamp方法获取时间戳，并维护一个最大时间戳。通过ExecutionConfig.setAutoWatermarkInterval(...)定义生成watermark的间隔（每n毫秒）。根据这个间隔，周期性调用分配器的getCurrentWatermark()方法，为watermark分配值。

在flink自带的BoundedOutOfOrdernessGenerator分配器中， getCurrentWatermark是定期将当前watermark更新为最大时间戳减去允许延迟时间的值。

以下是两个使用AssignerWithPeriodicWatermarks 生成的时间戳分配器的简单示例：

/*** This generator generates watermarks assuming that elements arrive out of order,* but only to a certain degree. The latest elements for a certain timestamp t will arrive* at most n milliseconds after the earliest elements for timestamp t.*/
class BoundedOutOfOrdernessGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {val maxOutOfOrderness = 3500L // 3.5 secondsvar currentMaxTimestamp: Long = _override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {val timestamp = element.getCreationTime()currentMaxTimestamp = max(timestamp, currentMaxTimestamp)timestamp}override def getCurrentWatermark(): Watermark = {// return the watermark as current highest timestamp minus the out-of-orderness boundnew Watermark(currentMaxTimestamp - maxOutOfOrderness)}
}/*** This generator generates watermarks that are lagging behind processing time by a fixed amount.* It assumes that elements arrive in Flink after a bounded delay.*/
class TimeLagWatermarkGenerator extends AssignerWithPeriodicWatermarks[MyEvent] {val maxTimeLag = 5000L // 5 secondsoverride def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {element.getCreationTime}override def getCurrentWatermark(): Watermark = {// return the watermark as current time minus the maximum time lagnew Watermark(System.currentTimeMillis() - maxTimeLag)}
}

AssignerWithPunctuatedWatermarks

根据每个元素的event time生成watermark，通过extractTimestamp(...)方法为元素分配时间戳，通过checkAndGetNextWatermark(...)检查元素的watermark并更新watermark。

checkAndGetNextWatermark(...)方法的第二个参数是extractTimestamp(...) 返回的时间戳，根据这个时间戳决定是否要生成watermark。每当checkAndGetNextWatermark(...) 方法返回一个非空watermark，并且该watermark大于上一个watermark时，就会更新watermark。

class PunctuatedAssigner extends AssignerWithPunctuatedWatermarks[MyEvent] {override def extractTimestamp(element: MyEvent, previousElementTimestamp: Long): Long = {element.getCreationTime}override def checkAndGetNextWatermark(lastElement: MyEvent, extractedTimestamp: Long): Watermark = {if (lastElement.hasWatermarkMarker()) new Watermark(extractedTimestamp) else null}
}