Streaming Systems: Watermark

绝地灬酷狼 2023-06-20 06:09 6阅读 0赞

扫码关注公众号免费阅读全文：**冰山烈焰的黑板报**  
![在这里插入图片描述][20200906232656299.jpg_pic_center]

# 1. Definition #

考虑到数据是连续不断地被系统处理和产出，我们无疑期望知道 Event Time 窗口何时关闭才是安全。换言之，窗口不会再有其他更多的数据到来。所以，我们寄希望描述系统相对于它的无界输入的进度。

为了实现这一想法，我们对于流式数据做了一个基础的假设：每一条消息都有一个相关的逻辑事件时间戳（Event Time）。这个假设在连续的无界数据的上下文中是合理的，因为这意味着输入数据是连续生成。 大多数情况下，我们可以将原始事件发生的时间作为其逻辑事件时间戳。对于包含事件时间戳的所有输入消息，我们可以观察此类时间戳在任何 Pipeline 中的分布。 这样的 Pipeline 可能会分布在多个分区上并行处理并消耗输入消息，而不能保证各个分区之间的排序。 因此，此 Pipeline 中正在处理（In-Flight）的消息的事件时间戳集将形成分布，如图 1 所示。

消息被 Pipeline 接收，处理，最终完成处理。因此，每一条信息同一时刻都处于下面两种状态之一：

1.  ***In-Flight***：表示消息已经被接收，但是还尚未处理完成。
2.  ***Completed***：代表不需要对该消息再进行更多的处理。

![图 1 In-Flight 和 Completed 消息][1 In-Flight _ Completed]

随着时间的推移，越来越多的消息会被添加到 ***In-Flight*** 区域，同时越来越多 ***In-Flight*** 区域的消息会移动到 ***Completed*** 区域。这个分布有一个关键点，位于 ***In-Flight*** 区域的最左边，对应于我们 Pipeline 中任何未处理消息的最早的事件时间戳。于是，我们可以使用这个值来定义 **Watermark**：

> **Watermark 是尚未 *Completed* 的最早数据的单调递增时间戳。**

*注意*：如果只考虑最早的事件时间，则 **Watermark** 并不总是单调的，因为我们没有对输入进行任何假设。

**Watermark** 的这种定义包含了两个特性：

1.  ***Completeness（Watermark 通过 Window 的末端时）*** ：如果 **Watermark** 已超过某个时间戳 T，我们将通过其单调属性来保证在 T 或 T 之前的准时（非延迟数据）事件将不再进行任何处理。因此，可以正确地在 T 或 T 之前进行任何聚合操作。换句话说，**Watermark** 使我们知道何时关闭窗口是正确的。
2.  ***Visibility***：如果由于某些原因数据阻塞在 Pipeline 中，则 **Watermark** 就无法前进。

# 2. Watermark Creation #

为了能给数据源创建一个 **Watermark**，我们必须给该数据源分配一个逻辑事件时间。[Streaming Systems: the What, Where, When and How][Streaming Systems_ the What_ Where_ When and How] 一文介绍了 **Watermark** 无外乎两种：*Perfect* 或 *Heuristic*。下图展示了这两种 Watermark 之间的不同。  
![图 2 左侧 Perfect Watermark 和右侧 Heuristic Watermark 开窗求和][2 _ Perfect Watermark _ Heuristic Watermark]  
请注意，二者的区别性特征是 **Perfect Watermark** 确保水印包含所有数据，而 **Heuristic Watermark** 则允许丢弃一些延迟的数据。

在将 **Watermark** 创建为 **Perfect Watermark** 或者 **Heuristic Watermark** 之后，**Watermark** 在系统的剩余部分都将保持不变。那么，是什么决定了应该创建 **Perfect Watermark**，还是 **Heuristic Watermark** 呢？这很大程度上取决于所消费的数据源的性质。

## 2.1 Perfect Watermark Creation ##

**Perfect Watermark** 的创建将时间戳分配给输入数据，以使最终的 **Watermark** 严格的保证不会再从此源再次看到事件时间小于 **Watermark** 的数据。使用 **Perfect Watermark** 的系统永远不必处理延迟数据。

但是，**Perfect Watermark** 的创建需要对输入数据有全面的了解，因此对于许多现实世界中的分布式输入源来说是不切实际的。这里有两个可以使用 **Perfect Watermark** 的例子：

1.  *Ingress Time（摄入时间）*：将进入时间指定为数据进入 Pipeline 的事件时间的输入源，可以创建一个 **Perfect Watermark**。
2.  *Static Sets of Time-Ordered Log（时间顺序日志的静态集合）*：时间顺序日志的静态大小的输入源（例如，具有静态分区集合的 Apache Kafka Topic，它的每个分区包含单调增加的事件时间）将是最简单的数据源，可以创建 **Perfect Watermark**。

## 2.2 Heuristic Watermark Creation ##

**Heuristic Watermark** 仅仅是估计不会再收到事件时间小于 Watermark 的数据。 使用 **Heuristic Watermark** 的 Pipeline 可能需要处理一些延迟数据。 延迟数据（Late Data）是 Watermark 大于该数据的事件时间之后到达的数据。延迟数据（Late Data）仅在 **Heuristic Watermark** 创建时才有可能。如果 Heuristic 是设置的很合理，延迟数据的数量可能非常小，并且 **Watermark** 作为完整性的估计仍然有用。 如果要支持需要正确性的用例（例如，计费之类的事情），系统仍然需要为用户提供一种处理延迟数据的方法。

对于许多现实中的分布式输入源，构建 **Perfect Watermark** 在计算或操作上是不切实际的，但仍然可以通过利用输入数据源的结构特征来构建高度准确的 **Heuristic Watermark**。

对于 **Perfect Watermark** 的创建，从广义上讲，对数据源的了解越多，启发式效果越好，并且将看到的延迟数据越少。 这里没有一刀切的解决方案，因为数据源的类型、事件的分布和使用模式会有很大差异。Watermark 的好处是：你可以将处理整个 Pipeline 的完整性问题，简化为在数据源创建 **Watermark**。

# 3. Watermark Propagation #

到目前为止，我们讨论了 **Watermark** 对于单个算子或 Stage 输入的情况。但是，现实中会存在由多个 Stage 组成的 Pipeline。因此，了解 Watermark 如何在独立 Stage 间传播对于了解它们如何影响整个 Pipeline 以及观察到的结果延迟非常重要。

我们可以在 Pipeline 中的任何单个算子或 Stage 的边界处定义 **Watermark**。 这不仅有助于了解 Pipeline 中每个 Stage 的相对进度，而且有助于为每个 Stage 尽快地及时发送结果。下面是 **Watermark** 在 Stage 边界的定义：

*  **Input Watermark**：**Input Watermark** 可以获取该 Stage 上游的所有进度。
    
     *  对于 *Source Upstream*，**Input Watermark** 是为输入数据创建 Watermark 的特定功能；
     *  对于 *Nonsource Upstream*，**Input Watermark** 被定义为其上游的所有 shards/partitions/instances 的 **Output Watermark** 的最小值。
 *  **Output Watermark**：**Output Watermark** 获取该 Stage 自身的进度，它被定义为该Stage 的 **Input Watermark** 和该 Stage 内所有未延迟数据的 Event Time 的最小值。

定义特定 Stage 的 **Input Watermark** 和 **Output Watermark** 的有一个不错的特性：我们可以使用它们来计算某个 Stage 所引入的 Event Time 延迟量。从其 **Input Watermark** 的值中减去该 Stage 的 **Output Watermark** 的值，即可得出该 Stage 引入的 Event Time 延迟或滞后。滞后是指每个 Stage 的输出滞后于实时的概念。例如，执行 10 秒窗口聚合的 Stage 将有 10 秒或更长时间的延迟，这意味着 Stage 的输出将至少比输入和实时延迟这么多。**Input Watermark** 和 **Output Watermark** 的定义提供了贯穿 Pipeline 的 Watermark 递归关系。

每个 Stage 中的处理也不是单一的。我们可以在一个 Stage 内将处理分成具有多个组件的流程，每个组件对输出 **Watermark** 有贡献。从概念上讲，每个这样的组件都充当一个缓冲区，活动消息可以驻留在其中，直到某个操作完成。例如，当数据到达时，它被缓冲以进行处理。然后处理可能会将数据写入状态，以便稍后进行延迟聚合。当触发延迟聚合时，可能会将结果写入等待下游消耗的输出缓冲区。

我们可以用它自己的 **Watermark** 跟踪每个这样的缓冲区。每个 Stage 缓冲器上的最小 Watermark 构成该 Stage 的输出水印。**Output Watermark** 可能是以下项中的最小值：

*  Per-source Watermark——对于每个发送 Stage。
 *  Per-external Input Watermark——对于 Pipeline 的外部数据源。
 *  Per-state Component Watermark——对于可以写入的每种类型的状态。
 *  Per-output Buffer Watermark——对于每个接收 Stage。

## 3.1 Watermark Propagation and Output Timestamp ##

由于 Watermark 是单调递增的，在这种限制下，可以推断出指定窗口的有效时间戳范围始于该窗口中最早的未延迟数据的时间戳。但是，大多数情况下，只有几种选择：

1.  *End of the window*：如果希望输出时间戳代表窗口边界，则使用窗口结束时间是最好的选择。

> *Start of The Window*：从 Watermark 正确性来看，窗口的起始端不是安全的选择。因为窗口中的第一个元素的时间戳通常晚于窗口本身的开始时间戳，这意味着不能保证 Watermark 一直被阻止到窗户开始。

1.  *Timestamp of First Nonlate Element*：当希望使用的 Watermark 尽可能保守时，用第一个未延迟元素的时间戳是一个不错的选择。但是，要权衡的是 Watermark 进度可能会受到更大的阻碍。
2.  *Timestamp of A Specific Element*：对于某些案例，使用某一元素的时间戳（从系统层面看）会是正确的选择。

现在我们用个例子来看看输出时间戳在整个 Pipeline 中的作用。我们将使用窗口的最早时间戳：第一个非延迟元素的时间戳作为窗口的时间戳。  
![图 3 使用窗口的最早时间戳作为输出时间戳][3]  
为了说明输出时间戳选择的效果，第一 Stage 中的虚线，显示每个 Stage 的 **Output Watermark** 所保持的内容。 与下图相比，输出 **Watermark** 被我们选择的时间戳延迟，其中输出时间戳被选择为窗口的结束。 从这个图中可以看出，第二 Stage 的输入 **Watermark** 也因此被延迟了。  
![图 4 输出时间戳被选择为窗口的结束][4]  
上图左侧中的 **Watermark** 对应于会话窗口末尾的输出时间戳。右侧 **Watermark** 位于会话窗口的开始处。我们可以看到图中水印线的延时更大，得到的平均会话长度不同。通过对比，有两点需要注意的地方：  
4. *Watermark 延迟*：第二种方式 **Watermark** 明显比第一种方式慢，因为只有窗口结束，**Watermark** 才能向前推进。  
5. *语义不同*：因为会话时间戳被分配到匹配会话中最早的非延迟元素，而各个会话通常会在不同的固定窗口中结束。到目前为止，我们所看到的这两种选择并没有本质上的对错之分，他们只是不同。但重要的是要理解它们是原理，才能准确地使用它们。

## 3.2 百分位 Watermark ##

截至目前为止，我们关注的都是通过一个 Stage 中活动消息的最小 Event Time 来衡量的Watermark。 这种方式可以使系统知道所有较早的时间戳何时已被考虑在内。 另一方面，我们可以考虑活动消息的事件时间戳的整个分布，并利用它来创建更细粒度的触发条件。

除了分布的最小值，我们还可以去分布的任意百分位数。如果对于业务逻辑来说“大部分”正确就足够了，那么百分位 Watermark 提供了一种机制，通过这种机制，Watermark 可以比我们通过从 Watermark 中丢弃分布的长尾中的异常值来跟踪最小事件时间更快更平稳地推进。  
![图 5 普通 Watermark 直方图][5 _ Watermark]  
![图 6 有异常值的 Watermark 直方图][6 _ Watermark]  
图 5 中 Event Time 分布中 90% 处的百分位 Watermark 和 100% 处的百分位 Watermark 是紧接着的。而图 6 中 90% 处的百分位 Watermark 和 100% 处的百分位 Watermark 有一段间隔。90% 的 Watermark 明显领先于 100% 的 Watermark。 通过从 Watermark 中丢弃异常数据，百分位 Watermark 仍然可以跟踪大部分分布，而不会被异常数据所延迟。  
![图 7 不同百分位 Watermark 的效果][7 _ Watermark]  
图 7 中演示了 33%、66%、100% 三种百分位 Watermark 跟踪数据分布中相应的时间戳百分比。注意，33% 和 66% 的 Watermark 会使窗口更早输出，当然后果是更多的数据成了延迟数据。例如，\[12:00,12:02) 的窗口中，33% 的 Watermark，只有 4 个元素会被计算，其他元素都是延迟数据，但是窗口在处理时间的 12:06 就会输出。66% 的 Watermark 有 7 个元素，在12:07 输出，而 100% Watermark 有 10 个元素，窗口要在 12:08 才会输出。

## 3.3 Processing-Time Watermark ##

Event-Time Watermark 不足以区分旧数据和系统延迟。换言之，Event-Time Watermark 无法区分正在快速且无延迟地处理一个小时前的数据的系统和试图处理被延迟了一个小时的实时数据的系统，即数据延迟（Data Latency）和系统延迟（System Latency，如，网络传输导致的系统延迟）。  
![图 8 Event-Time Watermark][8 Event-Time Watermark]  
图 8 中 Event-Time Watermark 延迟正在增加。但是，从这个信息中我们无法辨别是由于数据问题导致的，还是系统问题导致的。

![图 9 Processing-Time Watermark][9 Processing-Time Watermark]  
结合图 8 Event-Time Watermark 延迟和图 9 Processing-Time Watermark 延迟，我们可以推断出由于某种原因导致系统无法及时处理数据，比如网络问题等等。Processing-Time 上的延迟表明了由于某些系统异常导致了数据无法被及时处理，需要引起管理员的注意。我们接下来再看两种情况。  
![图 10 Event-Time Watermark 延迟增加而 Processing-Time Watermark 正常][10 Event-Time Watermark _ Processing-Time Watermark]  
图 10 中 Event-Time Watermark 延迟增加而 Processing-Time Watermark 正常，这表示数据在系统中缓存并等待处理，而不是系统操作阻止数据处理完成。  
![图 11 固定窗口中 Event-Time Watermark 延迟增加而 Processing-Time Watermark 正常][11 _ Event-Time Watermark _ Processing-Time Watermark]  
图 11 中 Event-Time Watermark 延迟随着每个窗口的元素被缓冲而增加，并随着每个窗口的聚合被触发而减少，而 Processing-Time Watermark 只是跟踪系统级延迟（在健康的情况下保持相对稳定）。

Processing-Time Watermark 可以有效地区分数据延迟和系统延迟。除了可见性之外，我们还可以在系统实现级别上使用 Processing-Time Watermark，用于诸如临时状态的垃圾收集这样的任务。

# 4. 总结 #

本文详细介绍了 Watermark 的定义、Perfect Watermark 和 Heuristic Watermark 的创建、Watermark 的传播和选择，以及 Event-Time Watermark 和 Processing-Time Watermark 的对比。

[20200906232656299.jpg_pic_center]: https://img-blog.csdnimg.cn/20200906232656299.jpg#pic_center
[1 In-Flight _ Completed]: https://img-blog.csdnimg.cn/59e920b04bf64d9d82951dbbc9906257.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[Streaming Systems_ the What_ Where_ When and How]: https://mp.weixin.qq.com/s/_l5Bj-hJv0fb6hZC-lVCAg
[2 _ Perfect Watermark _ Heuristic Watermark]: https://img-blog.csdnimg.cn/fd0e72b316164c1cb57b3f82b527683a.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[3]: https://img-blog.csdnimg.cn/be626e35a65f4de1b2ddb024c8babe8e.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[4]: https://img-blog.csdnimg.cn/a10bcde195ae4ee3bf86837d4611f62f.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[5 _ Watermark]: https://img-blog.csdnimg.cn/bba48d9d337740819b9db6e4603401cb.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[6 _ Watermark]: https://img-blog.csdnimg.cn/150e58ab9dfb4f6394e28c6a2aa898b8.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[7 _ Watermark]: https://img-blog.csdnimg.cn/3a05b9da96904507837ebf2c51903756.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[8 Event-Time Watermark]: https://img-blog.csdnimg.cn/0c4875e2feb44305acbec0b8a1ddc7b9.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[9 Processing-Time Watermark]: https://img-blog.csdnimg.cn/bea268184bc1462d81b03c4c246dbc0f.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[10 Event-Time Watermark _ Processing-Time Watermark]: https://img-blog.csdnimg.cn/e54f6af3ba7148298cc9aad165dfc036.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center
[11 _ Event-Time Watermark _ Processing-Time Watermark]: https://img-blog.csdnimg.cn/049d149dc6ca46828e6c18933180be6f.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBAR2F0c2J5TmV3dG9u,size_20,color_FFFFFF,t_70,g_se,x_16#pic_center