扫码关注公众号免费阅读全文：冰山烈焰的黑板报

1. Roadmap

Streaming 的数据流模型一共以下 5 个非常重要的概念，其中前两个在 Streaming Systems: Concept 中我们探讨过了，这里就不再做赘述。

1. Time Domain：

   • Event Time
   • Processing Time

2. Windowing

   • 滚动窗口（Tumbling Window）
   • 滑动窗口（Sliding Window）
   • 会话窗口（Session Window）
3. Trigger（触发器）
   - Trigger 是一种声明窗口何时应该输出的机制。从某种意义上说，可以将它视为一种流控制机制，用于指示何时输出结果。
   - Trigger 还可以在窗口演进时多次观察其输出。这就为随着时间推移更新结果提供了可能，它允许在数据到达时提供推测性的结果，以及处理上游数据随时间推移产生的变更或延迟到达的数据。
4. Watermark：Watermark 是一个关于 Event Time 的输入完整性的概念。一个时间 X 的 Watermark 表示：所有早于 X 的 Event Time 的数据都已经被观察到了。当观察一个没有已知终点的无界数据源时，Watermark 充当进度的度量。
5. Accumulation：Accumulation 指定针对同一个窗口观察到的多个结果之间的关系。这些结果可能是完全独立的，也可能是相互重叠的。不同的 Accumulation 模式在语义和处理成本上都是不同的，要根据具体场景来选择合适的 Accumulation 模式。

接下来的四个问题可以帮助我们更容易地理解这些概念之间的关系，而且它们是对于每个无界数据处理问题都是至关重要的：

1. What results are calculated？（计算什么结果）：这个问题是指选择 Transformation 的类型，比如求和、训练机器学习模型等。这也是批处理中的一个基本问题。
2. Where in event time are results calculated？（在事件时间中的哪个地方计算结果）：这个问题的答案就是 Event-Time Window，它包括我们之前介绍的滑动窗口、滚动窗口、会话窗口。当然，对于 Streaming Systems: Concept 一文中介绍的时间无关和近似算法这类无窗口化处理是没有窗口的概念的。另外，当把数据处理的时间当做 Event Time 时，它还包括 Processing-Time Window。
3. When in processing time are results materialized？（在什么处理时间点输出结果）：该问题实质上是指使用 Trigger 和 Watermark。虽然这个问题有很多种情况，但最常遇到的是重复更新的场景，即在相应的输入被确认是完整的之后，利用 Watermark 为窗口提供一个唯一的输出。

4. How do refinements of results relate？（如何更新相关的结果）：此问题本质上是选择 Accumulation：

   • Discarding：结果彼此之间相互独立；
   • Accumulating：后来的结果建立在之前的结果之上；
   • Accumulating and Retracting：累积值和先前的结果都被生产。

2. Batch：What and Where

2.1 What：Transformations

Transformation 也被应用于批处理。尽管您可能已经熟悉经典的批处理，但我们还是要从这里开始，因为它是我们添加所有其他概念的基础。如果你对 Spark 或者 Flink 有一定的了解，那么下面这几种 Transformation 也能很快理解。

比如，图 2 中对输入的数据进行求和，其中 X 轴表示 Event Time，Y 轴表示 Processing Time。系统实时的求和处理是从底部到顶部进行的，由随着 Processing Time 轴上的水平黑线随着时间推进所表示。输入是圆圈，圆圈中的数字是记录的值，他们开始是浅色的，随着实时处理的进行逐渐变暗。

当系统观察值时，将它们累积在中间状态，并最终将求和的结果物化输出。状态和输出由矩形表示（灰色表示状态，蓝色表示输出），聚合值在矩形顶部，矩形覆盖的区域表示事件时间和处理时间累积到结果中的部分。

对于批处理，它会累计状态，直到观察到所有输入（如图中绿色实线所示），然后产生最终的单个输出 48。在这个例子中，我们正在计算所有 Event Time 的总和，因为我们没有应用任何特定的窗口转换，因此状态和输出的矩形覆盖了 X 轴的整个部分。然而，如果我们想处理一个无界数据源，那么批处理是不够的，我们不能等待输入结束，因为它实际上永远不会结束。这时，你一定会想到我们之前的文章提到的窗口。到此，也就引出了第二个问题：Where（在事件时间中的哪个地方计算结果）。

2.2 Where: Windowing

在 Streaming Systems: Concept 一文中，我们知道开窗是沿时间边界对数据源进行切片的过程。一般来说，有三种窗口：固定窗口（Fixed Window）、滑动窗口（Sliding Window）和会话窗口（Session Window）。

为了更深入地理解窗口，我们现在以在 2 分钟固定窗口求和为例，观察系统是如何在窗口中进行求和计算的。

和之前一样，输入会累计到状态中，直到所有输入被消费完，然后产生输出结果。但是，这里我们将会得到四个输出结果，而不是一个，其中四个 2 分钟 Event Time 窗口会各产生一个输出结果。

在此，我们再次回顾了 Streaming Systems: Concept 中提到的 Event Time、Processing Time 和窗口。如若我们想进一步钻研，那么就不得不学习 Trigger、Watermark 和 Accumulation。

3. Streaming: When and How

3.1 When: Trigger

Trigger 提供了 When（在事件时间中的哪个地方计算结果）的答案。Trigger 声明在 Processing Time 内窗口的何时进行输出。窗口的每个特定输出都被称为窗口的一个 Pane（窗格）。（注意：Pane，这个概念虽然在 Spark 和 Flink 中没有明确定义，但也包含在 Trigger 中）。

虽然可以想象相当多种可能的 Trigger 语义，但从概念上讲，只有两种通常有用的触发器类型。在实际应用中，使用到的 Trigger 基本有两种，或者两种结合使用：

• Repeated Update Trigger：随着窗口内容的发展，它们定期地为窗口生成更新的 Pane。这些更新可以随每条新记录一起物化，也可以在某些处理时间延迟之后发生，如每分钟。这种触发周期的选择主要是平衡延迟和成本。
• Completeness Trigger：只有当窗口的输入被认为达到某个阈值时，才会物化窗口的 Pane。这种类型的 Trigger 与批处理最相似：只有在输入完成之后，才提供结果。不同之处是：这种基于 Trigger 的方法在于完整性的概念限定在单个窗口，而不是绑定于整个输入的完整性。

Repeated Update Trigger 是流式系统中最常见的 Trigger。它们实现简单且易于理解，常用于对物化数据集的重复（最终是一致的）更新，类似于数据库世界中物化视图的语义。Completeness Trigger 不太常见，但却提供的流语义与批处理的语义更接近，它们还提供了用于推理诸如丢失数据和延迟数据之类的工具，在后续关于 Watermark 的文章中为进一步讨论。

在此之前，我们先看一个 Repeated Update Trigger 的简单例子：每当遇到一条记录是进行触发的 Trigger。

我们可以看到图 5 中每个窗口会有多个输出（每个 Pane 一个）：一旦有输入就有输出。Per-Record Trigger 的缺点是它太频繁了。在处理大规模数据时，像求和这样的聚合为减少流的基数而不丢失信息提供了很好的机会。但是，有些时候，可能更愿意在某些处理时间延迟（比如每一秒或每一分钟）之后更新。

Trigger 在处理时间延迟上有两种不同的方法:

• Aligned Delays：这种延迟是指将处理时间分割成固定的区域，这些区域跨越键和窗口进行对齐。
• Unaligned Delays：这种延迟与给定窗口内观察到的数据有关（如，进入窗口的时间）。

如图 6 中，Aligned Delay Trigger 类似于微批处理（比如，Spark Streaming），它的优缺点如下：

• 优点

  • 它具有可预测性（如，可预测何时发生）；
  • 可以在同一时间获得所有修改窗口的定期更新。

• 缺点

  • 所有更新都同时发生，这会导致工作负载激增，通常需要更大的峰值供应才能正确处理负载。

图 7 中的 Unaligned Delay Trigger 与图 6 中的 Aligned Delay Trigger 正好相反。显而易见，Unaligned Delay Trigger 在时间上更均匀地分布负载。 任何给定窗口所涉及的实际延迟在两者之间有所不同，有时更多有时更少，但最终平均延迟将保持基本相同。因此，Unaligned Delay Trigger 通常是大规模数据处理的更好选择，因为它们会是负载分布的更均匀。

虽然 Repeated Update Trigger 可以周期性的更新结果，并且随着时间推移会朝着正确性收敛。但是，分布式系统的变化无常通常会导致事件发生时间和系统实际处理的时间之间存在不同程度的偏差，这意味着很难推断输出什么时候能显示输入数据的准确和完整视图。对于输入完整性很重要的情况，重要的是要有某种方法对完整性进行推理，而不是盲目相信由碰巧找到系统的数据子集所计算的结果。这就需要一个推测工具——Watermark。

3.2 When: Watermark

Watermark 是 Event Time 时间域中输入完整性的时间概念。 换句话说，它们是系统测量相对于事件流中正在处理的记录的 Event Time 的进度和完整性的方式（无论有界或无界，尽管它们的用处在无界情况下更为明显）。

对于大多数现实世界的分布式数据处理系统，事件时间和处理时间之间的偏差可以描述为一个不断变化的时间函数。

图中那条弯弯曲曲的红线本质上就是 Watermark。Watermark 捕获随着处理时间的推移而发生的事件时间完整性的进展。从概念上讲，可以把 Watermark 看作是一个函数 F ( P ) → E F(P) \rightarrow E F(P)→E，它的从 Processing Time 上取一个点，返回对应的 Event Time 的一个点。Event Time 的 E 点，表示系统认为所有 Event Time 小于 E 的输入都已经被发现了。换言之，它断言再也没有 Event Time 小于 E 的数据了。Watermark 有两种类型：

1. Perfect Watermark：在这种情景下，没有延迟的数据，所有的数据都准时到达。
2. Heuristic Watermark：Heuristic Watermark 利用输入所有的可用信息（如，分区、分区内的顺序、文件增长率等），提供尽可能准确的进度估计。在许多情况下，这样的 Watermark 可以非常准确的预测。即便如此，Heuristic Watermark 的使用意味着它有时可能是错误的，这将导致延迟数据。

因为 Watermark 提供了相对于输入的完整性概念，所以 Watermark 构成了前面提到的第二种触发器的基础：Completeness Trigger。由于 Watermark 本身比较复杂，我会另起一文进行讨论。

构建 Perfect Watermark 是一种理想情况，因为我们缺乏关于输入的全部信息，或者需要耗费巨大的计算成本才能算出 Perfect Watermark。此时，我们可能会使用 Heuristic Watermark。为了便于理解指定的输入集可以应用不同的水印这一想法，让我们看一个示例，它是在同一个数据集上执行的，但使用了两种不同的水印实现：左侧是 Perfect Watermark，右侧是 Heuristic Watermark。

正如所料，Perfect Watermark 完美地捕捉了系统随着时间推移的 Event Time 完整性。 相比之下，右侧 Heuristic Watermark 使用的特定算法没有考虑到 9 这个值，这在输出延迟和正确性方面彻底改变了物化输出的形状（如 [12:00, 12:02) 窗口提供了错误的结果 5）。

尽管如此，这个例子还是突出了 Watermark 两个非常明显的缺陷：

1. Too Slow：当任何类型的 Watermark 由于已知的未处理而正确地延迟的数据（由于网络带宽限制，导致输入增长变缓）时，如果你仅依赖 Watermark 的前进来触发结果计算，则直接转换为输出延迟。很重要的一点是：虽然 Watermark 提供了一个非常有用的完整性概念，但是从延迟的角度来看，依赖完整性来产生输出通常是不理想的。如图左侧的例子中，即迟到的 9 这个记录阻止所有后续窗口的 Watermark，即使这些窗口的输入数据已经更早的达到并且是完整的了。第二个窗口 [12:02, 12:04) 尤其明显，从窗口中的第一个值到达到窗口计算并输出结果的时间需要将近 7 分钟。这个例子中的 Heuristic Watermark 要稍微好一点（5分钟就输出了），但这不意味着 Heuristic Watermark 从来不会受到其他 Watermark 滞后的影响。
2. Too Fast：当 Heuristic Watermark 提前的时间不正确时，在 Watermark 之前具有 Event Time 的数据可能会延迟一段时间到达，从而导致了延迟数据。Heuristic Watermark 毕竟是 Heuristic，意味着本质上有时也会出错。因此，如果关心正确性，那么仅仅依靠 Heuristic Watermark 来确定何时物化输出是不够的。如图右侧的例子中，第一个窗口 [12:00, 12:02) 输入数据全部到达之前，Watermark 进入 [12:00, 12:02) 窗口的末尾，导致错误的输出值为 5 而不是 14。

由于 Watermark 这两个缺陷，完全依赖于完整性概念的系统无法同时获得低延迟和正确性。如果 Repeated Update Trigger 提供了低延迟更新，但无法推断完整性，而 Watermark 提供了完整性的概念，但可变且可能高延迟，那么为什么不将它们结合起来呢?

3.3 When: Early/On-Time/Late Trigger

我们现在已经了解了两种主要类型的触发器：Repeated Update Trigger 和 Completeness/Watermark Trigger。 在很多情况下，单独使用它们都不够，但将它们组合在一起就足够了。复合触发器具体化的窗格分为三类：

1. Zero or More Early Pane：它是 Repeated Update Trigger 的结果，该 Trigger 定期触发，直到 Watermark 通过窗口的末端。这些触发产生的 Pane 包含推测性结果，但允许我们观察新输入数据到达时窗口随时间的演变。这弥补了有时水印的 Too Slow 缺点。
2. A Single On-Time Pane：它是 Watermark 经过窗口末端后，触发Completeness/Watermark Trigger 的结果。它使系统现在相信这个窗口的输入是完整的。这意味着现在可以安全地推算出丢失的数据；例如，在执行外部连接时提交部分连接。
3. Zero or More Late Pane：它是另一个（可能不同的）Repeated Update Trigger 的结果，该 Trigger 在 Watermark 经过窗口末端后，每当延迟数据到达时，都会周期性地触发。在 Perfect Watermark 的情况下，总是没有延迟的 Pane。但在 Heuristic Watermark 的情况下，Watermark 未能正确处理的任何数据都将导致延迟触发。这弥补了 Watermark 的 Too Fast 缺点。

现在我们再看一个例子，使用一个基于 Processing Time 的周期 Trigger，对早期触发使用一分钟的延迟，对后期触发使用 Per-Record Trigger。这样，早期的触发将为大容量窗口提供一定数量的批处理（由于触发器每分钟只触发一次，而不管窗口的吞吐量如何），但我们不会为后期的触发引入不必要的延迟，如果我们用的是合理准确的 Heuristic Watermark ，这种情况很少发生。

这个版本有两个明显的改进：

1. 对于第二个窗口 [12:02, 12:04) 中的 Watermark Too Slow 的情况：我们现在每分钟提供一次周期的提前计算。在 Perfect Watermark 情况下差异最为明显，首次输出时间从近 7 分钟减少到 3 分半； 但它在 Heuristic Watermark 情况下也明显改进。 现在，这两个版本都随着时间的推移提供稳定的改进（值为 7、10 和 18 的窗格），在输入完成和窗口最终输出 Pane 的具体化之间具有相对最小的延迟。
2. 对于第一个窗口 [12:00, 12:02) 中的 Heuristic Watermark Too Fast 情况：当 9 的记录出现较晚时，我们立即将其合并到一个新的更正的窗格中，其值为 14。

这些新触发器的一个有趣的副作用是，它们有效地标准化 Perfect Watermark 和 Heuristic Watermark 版本之间的输出模式。由于使用了水印触发器，我们还可以推断我们使用 Early/On-Time/Late Trigger 生成的结果中的输入完整性。 这使我们能够更好地处理关心缺失数据的用例，例如外连接、异常检测等。

在这一点上，Perfect 和 Heuristic Early/On-Time/Late 之间最大的剩余区别是窗口生命周期界限。 在 Perfect Watermark 的情况下，我们知道在 Watermark 结束后我们将再也看不到窗口的任何数据，因此我们可以删除当时窗口的所有状态。 在 Heuristic Watermark 的情况下，我们仍然需要在一段时间内保持窗口状态以考虑延迟数据。 但是到目前为止，我们的系统还没有任何好的方法来知道每个窗口需要保持多长时间的状态。 这就是允许迟到的地方。

3.4 When: Allowed Lateness

在继续我们的最后一个问题（结果如何修正？）之前，先聊一聊长期存在的无序流处理系统中的实际问题：垃圾收集。 在 Heuristic Watermark 示例中，每个窗口的持久状态在示例的整个生命周期内都存在； 这是必要的，以便我们能够当延迟数据到达时进行处理。 但是，虽然能够将所有持久状态保留到时间结束会很棒，但实际上，在处理无界数据源时，无限期地保留给定窗口的状态（包括元数据）通常是不切实际的，我们最终会耗尽磁盘空间。

因此，任何现实世界的乱序处理系统都需要提供某种方式来限制它正在处理的窗口的生命周期。 一种简洁明了的方法是定义系统内允许延迟的范围； 也就是说，对任何给定记录（相对于 Watermark）的延迟时间进行限制，以便系统处理它； 在此范围之后到达的任何数据都将被简单地丢弃。 在确定了单个数据的延迟时间后，还精确地确定了窗口状态必须保持多长时间：直到 Watermark 超过窗口结束的延迟时间范围。 但除此之外，还要赋予系统自由，可以在观察到后立即丢弃任何晚于地平线的数据，这意味着系统不会浪费资源处理没人关心的数据

因为 Allowed Lateness 和 Watermark 之间有点微妙，所以值得看一个例子。 我们并在图 7 中添加一分钟的延迟时间范围，在其中我添加了以下特性来突出允许延迟的效果：

1. 加粗的水平方向的黑线表示当前的 Processing Time，注释为 Lateness Horizon 的小刻度表示窗口的最大延时范围（事件时间）。
2. 当 Watermark 通过窗口的延迟时间范围时，该窗口将关闭，这意味着该窗口的所有状态都将被丢弃。 我留下一个虚线矩形，显示窗口关闭时所覆盖的时间范围，还有一条向右延伸的小尾巴，表示窗口的延迟视界（与 Watermark 形成对比）。
3. 同时，在第一个窗口添加了一个附加的延迟数据，值为 6。6 是延迟的，但仍在允许的延迟范围内，因此被合并到值为 11 的更新结果中。但是，9 到达迟到范围之外，因此它被简单地丢弃。

关于迟到期限的最后两点补充说明:

1. 如果使用的数据恰好来自可获得 Perfect Watermark 的数据源，则不需要处理延迟数据，允许的零秒延迟范围将是最优的。
2. 有一些例外情况不需要指定最大延迟，即使在使用 Heuristic Watermark 时，类似于计算所有时间的全局聚合以获得可处理的有限数量的 Key。 在这种情况下，系统中活动窗口的数量受到使用的有限 Key 的限制。 只要 Key 的数量保持在可管理的低水平，就无需担心通过允许的延迟来限制窗口的生命周期。

3.5 How: Accumulation

当使用 Trigger 在一段时间内为单个窗口生成多个 Pane 时，我们发现自己面临最后一个问题：（每个 Pane 的）结果如何修正。这有 4 种不同的 Accumulation 模式：

1. Discarding：每次实现 Pane 结果物化时，任何存储状态都将被丢弃。 这意味着每个连续的 Pane 都独立于之前的 Pane 。 当下游使用者自己进行某种积累时，Discarding 模式很有用。 例如，增量求和。
2. Accumulating：每次实现 Pane 结果物化时，任何存储的状态都会保留，并将将来的输入累积到现有状态中。 这意味着每个连续的 Pane 都建立在先前的 Pane 上。 当以后的结果可以简单地覆盖以前的结果时，Accumulating 模式很有用。

3. Accumulating and Retracting：这类似于 Accumulating 模式，但是在生成新 Pane 时，它还会为以前的 Pane 生成独立的 Retraction（回撤）。 Retraction（与新的累加结果结合使用）本质上的语义是：“我之前告诉过你结果是 X，但我错了。 抛弃我上次告诉你的 X，然后将其替换为 Y”。在两种情况下，Retraction 特别有用：

   1. Regrouping data by a different dimension：当下游按不同的维度对数据进行重新分组时，新值很可能最终以与先前值不同的 Key 输入，从而最终分布在不同的组。在这种情况下，新值不能只覆盖旧值；相反，需要回撤（Retraction）操作以删除旧值。
   2. Dynamic Window：当使用动态窗口（例如，Session Window）时，由于窗口合并，新值可能会替换多个以前的窗口。在这种情况下，可能难以仅从新窗确定要替换哪些旧窗口。 对旧窗口进行显式回撤（Retraction）可以使任务简单明了。
4. Discarding and Retracting (大多数情况下，这种模式不是非常有用)

当放在一起看时，各种累积模式的不同语义会更清晰一些。 考虑图 10 中第二个窗口的两个 Pane（事件时间范围为 [12:06, 12:08) 的 Pane）（带有 Early/On-Time/Late Trigger 的 Pane）。 下表显示了每个 Pane 的值在三种累积模式下的样子（累积模式是图 10 本身使用的特定模式）。

现在我们来看看发生了什么：

• Discarding：每个 Pane 仅包含在特定 Pane 中到达的值。因此，该窗口的最终值是最后一个窗格的值，而不是总和。但是，如果想要对所有独立窗格求和，将得到正确的答案 12。这就是为什么丢弃模式下，下游消费者自己在 Pane 上执行聚合时很有用。
• Accumulating：每个 Pane 都包含在该特定 Pane 期间到达的值，以及来自先前 Pane 的所有值。 因此，观察到的最终值是 12。 但是，如果您要对各个 Pane 本身求和，则实际上会重复计算来自 Pane 1 的输入，从而得出错误的总和 15。这就是为什么当可以用新值简单地覆盖以前的值时，累积模式最有用：新值已经包含了迄今为止看到的所有数据。
• Accumulating & Retracting：每个 Pane 都包括一个新的累积模式的值以及前一个 Pane 的回撤值。 因此，观察到的最后一个值（不包括回撤）以及所有实体化窗格的总和（包括回撤）都是 12。这就是回撤如此强大的原因。

在丢弃模式下，带有 Heuristic Watermark 的流式系统上再次运行将产生如图 12 所示。

尽管输出的整体形状类似于图 10 中的累积模式版本，但注意此丢弃版本中的所有 Pane 都没有重叠。 因此，每个输出都独立于其他输出。

我们在来看看累积回撤模式下会有哪些变化。

因为每个窗口的 Pane 都重叠，所以要清楚地看到回撤（Retracting）有点难。 回撤以红色表示，与重叠的蓝色 Pane 相结合，产生略带紫色的颜色。 在给定 Pane 中水平列出了两个输出的值（并用逗号将它们分开），以使它们更容易区分。

下图将图 8、图 9 和图 13 并列放在一起，可以清晰地看出三者异同。

可以看出，按照上图的顺序丢弃模式、累积模式、累积回撤模式在存储和计算成本方面变得越来越多。 为此，Accumulation 模式的选择为沿着正确性、延迟和成本轴进行权衡提供了另一个维度。

4. Conclusion

本文首先讲了 Event Time 和 Processing Time、Window、Trigger、Watermark 和 Accumulation，然后围绕着以下四个问题进行了更详细的阐释：

1. What results are calculated? = Transformations.
2. Where in event time are results calculated? = Windowing.
3. When in processing time are results materialized? = Triggers plus Watermarks.
4. How do refinements of results relates? = Accumulations.