目录#
到目前为止，我们已经从流处理的角度来看待管道作者或数据科学家。第2章将水位线引入作为回答事件时间处理正在进行的位置和处理时间结果何时实现的基本问题的一部分。在本章中，我们从流处理系统的基本机制的角度来看待相同的问题。观察这些机制将有助于我们激发、理解和应用水位线的概念。我们讨论了水位线如何在数据输入点创建、如何通过数据处理管道传播以及如何影响输出时间戳。我们还演示了水位线如何保留必要的保证，以回答事件时间数据在何处处理以及何时它被实现，同时处理无界数据。
定义#
考虑任何连续摄取数据并输出结果的管道。我们希望解决何时可以安全地关闭事件时间窗口的一般问题，即窗口不再期望有任何数据。为了解决这个问题，我们希望相对于无限输入特征化管道正在做的进展。
解决事件时间窗口问题的一种天真的方法是只基于当前处理时间来设置事件时间窗口。如第一章所述，我们很快就会遇到麻烦，因为数据处理和传输不是瞬间完成的，因此处理和事件时间几乎永远不会相等。我们的管道中的任何故障或峰值可能会导致我们错误地将消息分配到窗口中。最终，这种策略失败了，因为我们没有可靠的方法来保证这样的窗口。
另一种直观但最终不正确的方法是考虑管道处理的消息速率。虽然这是一个有趣的指标，但速率可能会因输入变化、预期结果的变化、可用于处理的资源等因素而任意变化。更重要的是，速率无法帮助回答完整性的根本问题。具体而言，速率无法告诉我们何时看到了特定时间间隔的所有消息。在现实世界的系统中，会有消息无法通过系统进行进展的情况。这可能是暂时性错误（例如崩溃、网络故障、机器停机）的结果，也可能是需要更改应用逻辑或其他手动干预的应用级故障的结果。当然，如果发生大量故障，则处理速率指标可能是检测到这种情况的良好代理。但是，速率指标永远无法告诉我们单个消息未能通过管道进行进展。然而，即使单个这样的消息，也可能会任意影响输出结果的正确性。
我们需要更可靠的进展度量。为了到达那里，我们对我们的流数据做出一个基本假设：每个消息都有一个关联的逻辑事件时间戳。在不断到达的无界数据的上下文中，这种假设是合理的，因为这意味着输入数据的持续生成。在大多数情况下，我们可以将原始事件发生的时间作为其逻辑事件时间戳。由于所有输入消息都包含事件时间戳，因此我们可以检查任何管道中这些时间戳的分布。这样的管道可能被分布在许多代理上并以并行方式处理，并且没有保证单个分片之间的顺序。因此，此管道中正在进行的活动消息的事件时间戳集将形成分布，如图3-1所示。
消息被管道摄取、处理，最终被标记为已完成。每个消息都是“运行中”的，表示已经接收但尚未完成，还是“已完成”的，表示不需要在此消息上执行更多处理。如果我们按事件时间检查消息的分布，它将看起来像图3-1。随着时间的推移，“运行中”分布的右侧将添加更多的消息，“运行中”部分的消息将被完成并移动到“已完成”分布。
图3-1.流水线中的运行中和完成的消息事件时间分布。新消息作为输入到达并保持“运行中”，直到为它们完成处理为止。在任何给定时刻，“运行中”分布的最左侧边缘对应于最旧的未处理元素。
在此分布中有一个关键点，位于“运行中”分布的最左边缘，对应于我们管道中任何未处理消息的最旧事件时间戳。我们使用此值来定义水位线：
水位线是最旧未完成的工作的单调递增时间戳。
此定义提供了两个基本属性，使其有用：
完整性
如果水位线已超过某个时间戳T，则由其单调属性保证，不会再发生处理，因此我们可以正确地发出任何T时间以及T之前的时段的聚合结果。换句话说，水位线使我们知道何时正确地关闭窗口。可见性
如果消息因任何原因在管道中卡住，则水位线无法前进。此外，我们将能够通过检查阻止水位线前进的消息来找到问题的来源。
源水位线的创建#
这些水位线从哪里来？为了为数据源建立水位线，我们必须为从该源进入管道的每个消息分配逻辑事件时间戳。正如第2章所述，所有水位线创建都属于以下两个广泛类别之一：完美或启发式。为了提醒我们完美和启发式水位线之间的区别，让我们看一下第2章中的窗口求和示例的图3-2。
图3-2。完美（左）和启发式（右）水位线的窗口求和
请注意，区别的特点在于完美的水位线确保水位线考虑到所有数据，而启发式水位线则允许一些后到数据元素。
水位线创建为完美或启发式后，水位线将在整个管道中保持不变。至于是什么使水位线创建完美还是启发式，这在很大程度上取决于正在使用的源的性质。为了看到原因，让我们看一下每种类型的水位线创建的一些示例。
完美水位线创建#
完美的水位线创建分配时间戳给进入的消息，以使得产生的水位线是严格保证，即从此源再也看不到事件时间小于水位线的数据了。使用完美水位线创建的管道永远不必处理后到数据；也就是说，在水位线超过新到达消息的事件时间之后到达的数据。然而，完美的水位线创建需要对输入有完美的了解，因此对于许多真实世界的分布式输入源来说是不切实际的。以下是可以创建完美水位线的用例的一些示例：
进入时间戳
将进入时间分配为进入系统的数据的事件时间的源可以创建完美的水位线。在这种情况下，源水位线仅跟踪流水线观察到的当前处理时间。这本质上是近年来支持窗口的几乎所有流处理系统使用的方法。
因为事件时间是从一个单一的、单调递增的源（实际处理时间）分配的，所以系统对于数据流中接下来的哪些时间戳将到来具有完美的了解。因此，事件时间进度和窗口语义变得容易推理。当然，缺点是水位线与数据本身的事件时间没有关联；这些事件时间被有效地丢弃，水位线只是跟踪数据相对于其到达系统的进度。
时间有序日志的静态集
时间有序日志的静态大小输入源（例如，具有静态分区集的Apache Kafka主题，其中源的每个分区包含单调递增的事件时间）将是一个相对简单的源，可以在其上创建完美的水位线。为此，源将简单地跟踪未处理数据的已知和静态源分区集的最小事件时间（即，每个分区中最近读取的记录的事件时间的最小值）。
类似于前面提到的进入时间戳，由于静态分区集中的事件时间已知单调递增，因此系统对于接下来哪些时间戳将到来具有完美的了解。这实际上是一种有界的无序处理形式；已知分区集中的无序程度由这些分区中观察到的最小事件时间所限制。
通常，您只能在分区内保证单调递增的时间戳，如果在写入数据时为这些分区分配时间戳，例如通过Web前端直接记录事件到Kafka中。尽管仍然是有限的用例，但这绝对比到达数据处理系统时的进入时间戳有用得多，因为水位线跟踪底层数据的有意义的事件时间。
启发式水位线创建#
另一方面，启发式水位线创建创建的水位线仅是估计，即事件时间小于水位线的数据将不会再次出现。使用启发式水位线创建的管道可能需要处理一些后到数据。后到数据是任何在水位线超过此数据的事件时间之后到达的数据。后到数据仅在使用启发式水位线创建时可能出现。如果启发式是一个合理好的启发式，那么后到数据的数量可能非常小，水位线仍然有用作为完成估计。如果系统要支持需要正确性的用例（例如，像计费这样的东西），则仍然需要提供一种处理后到数据的方法。
对于许多真实世界的分布式输入源来说，构建完美的水位线在计算或操作上是不切实际的，但是利用输入数据源的结构特征仍然可以构建高度准确的启发式水位线。以下是两个示例，其中可以使用启发式水位线（具有不同质量）：
动态时间有序日志集
考虑一组动态结构化日志文件（每个单独的文件包含相对于同一文件中的其他记录单调递增的事件时间，但事件时间之间没有固定的关系），其中预期的完整日志文件集（即，在Kafka术语中的分区）在运行时未知。这样的输入通常在由多个独立团队构建和管理的全球规模服务中发现。在这种情况下，创建输入的完美水位线是棘手的，但创建准确的启发式水位线是完全可能的。
通过跟踪现有日志文件中未处理数据的最小事件时间、监视增长率以及利用外部信息（如网络拓扑和带宽可用性），即使缺乏有关所有输入的完美知识，也可以创建一个非常准确的水位线。这种类型的输入源是在Google发现的最常见的无界数据集之一，因此我们在这些情况下有着丰富的创建和分析水位线质量的经验，并看到它们在许多用例中被很好地使用。
Google Cloud Pub/Sub
Cloud Pub/Sub是一个有趣的用例。Pub/Sub目前不保证按顺序传递；即使单个发布者按顺序发布两个消息，它们也有可能（通常很小）按顺序传递（这是由于底层架构的动态性，允许在零用户干预下扩展到非常高的吞吐量）。因此，无法保证Cloud Pub/Sub的完美水位线。然而，Cloud Dataflow团队通过利用有关Cloud Pub/Sub中数据的可用信息构建了一个相当准确的启发式水位线。本章后面将详细讨论此启发式的实现作为案例研究。
考虑一个用户玩手机游戏的示例，他们的分数被发送到我们的管道进行处理：通常可以假设对于任何使用移动设备作为输入的源，提供完美的水位线基本上是不可能的。由于设备长时间离线的问题，没有办法为这种数据源提供任何合理的完整性估计。但是，您可以想象构建一个准确地跟踪当前在线设备的输入完整性的水位线，类似于上面描述的Google Pub/Sub水位线。从提供低延迟结果的角度来看，处于活动状态的用户可能是最相关的用户子集，因此这通常不是您最初想到的缺点。
总的来说，启发式水位线的创建越了解源，启发式就越好，就越少见到后到数据项。鉴于源类型、事件分布和使用模式会有很大的变化，因此没有一种万能的解决方案。但是无论哪种情况（完美还是启发式），在输入源处创建水位线后，系统可以完美地传播水位线。这意味着完美水位线将在下游保持完美，启发式水位线将严格保持与创建时一样的启发式。这是水位线方法的好处：您可以将管道中的完整性跟踪复杂性完全减少到在源处创建水位线的问题上。
水位线传播#
到目前为止，我们仅考虑了单个操作或阶段内输入数据的水位线。然而，大多数实际的流水线都由多个阶段组成。了解水位线在独立阶段之间如何传播对于理解它们如何影响整个流水线以及其结果的观察延迟非常重要。
1 | **流水线阶段** |
水位线是在输入源处创建的，就像前面讨论的那样。然后在数据通过系统时，它们在概念上通过系统流动。您可以在不同粒度级别上跟踪水位线。对于由多个独立阶段组成的流水线，每个阶段可能都跟踪其自己的水位线，其值是其前面的所有输入和阶段的函数。因此，流水线后面的阶段将具有更过去的水位线（因为它们看到的整体输入要少）。
我们可以在流水线中任何单个操作或阶段的边界上定义水位线。这不仅有助于了解流水线中每个阶段的相对进度，而且对于独立地且尽快为每个单独的阶段分派及时结果非常有用。我们为阶段边界处的水位线给出以下定义：
- 输入水位线（input watermark）：捕获该阶段上游的所有进度（即该阶段的输入数据完整程度）。对于源，输入水位线是创建输入数据水位线的源特定函数。对于非源阶段，输入水位线定义为所有上游源和阶段的所有分片/分区/实例的输出水位线的最小值。
- 输出水位线（output watermark）：捕获阶段本身的进度，本质上定义为阶段的输入水位线和所有非延迟数据激活消息的事件时间的最小值。 “活动”确切包含的内容在很大程度上取决于给定阶段实际执行的操作以及流处理系统的实现。它通常包括缓冲以进行聚合但尚未向下游实现的数据，正在等待输出数据传输到下游阶段等。
定义特定阶段的输入和输出水位线的一个好处是我们可以使用它们计算阶段引入的事件时间延迟量。将阶段的输出水位线值减去其输入水位线值可得出阶段引入的事件时间延迟或滞后。此滞后是每个阶段输出相对于实际时间的延迟程度的概念。例如，执行10秒窗口聚合的阶段将具有10秒或更多的滞后，这意味着阶段的输出至少要比输入和实际时间延迟那么多。输入和输出水位线的定义提供了整个流水线中水位线的递归关系。流水线中的每个后续阶段都会根据阶段的事件时间滞后必要地延迟水位线。
每个阶段内的处理也不是单一的。我们可以将一个阶段内的处理划分为具有多个概念组件的流，每个组件都有助于生成输出水位线。如前所述，这些组件的确切性质取决于阶段执行的操作和系统的实现。从概念上讲，每个这样的组件都充当缓冲区，其中活动消息可以驻留，直到某个操作完成。例如，数据到达时，它会被缓冲以进行处理。然后，处理可能会将数据写入状态以进行延迟聚合。当触发延迟聚合时，它可能会将结果写入等待下游阶段消耗的输出缓冲区，如图3-3所示。
图3-3。流系统阶段的示例系统组件，包含正在进行的数据的缓冲区。每个都将有关联的水位线跟踪，阶段的整体输出水位线将是所有此类缓冲区的水位线的最小值。
我们可以使用自己的水位线跟踪每个这样的缓冲区。每个阶段缓冲区中的水位线的最小值形成该阶段的输出水位线。因此，输出水位线可以是以下内容的最小值：
- 每个发送阶段的每个源水位线。
- 每个外部输入水位线——用于流水线外部的源
- 每种类型的状态组件水位线——可以写入每种状态
- 每个接收阶段的水位线
在这个粒度级别上提供水位线也可以更好地了解系统的行为。水位线跟踪系统中各个缓冲区中的消息位置，从而更容易诊断卡住的问题。
了解水位线传播#
为了更好地了解输入和输出水位线之间的关系以及它们如何影响水位线传播，让我们看一个例子。我们考虑游戏得分，但不是计算团队得分的总和，而是试图衡量用户参与水平。我们将首先通过计算每个用户会话的长度来实现这一点，假设用户参与游戏的时间是他们享受游戏程度的合理代理。在回答我们的四个问题以计算会话长度一次后，我们将再次回答这些问题，以在固定时间段内计算平均会话长度。
为了使我们的示例更加有趣，假设我们正在使用两个数据集，一个用于移动得分，另一个用于控制台得分。我们希望通过整数求和在这两个独立数据集上并行执行相同的得分计算。一个管道正在计算在移动设备上玩游戏的用户的得分，而另一个管道则是针对在家庭游戏机上玩游戏的用户，可能是由于针对不同平台采用了不同的数据收集策略。重要的是，这两个阶段执行相同的操作，但是在不同的数据上执行，因此输出的水位线非常不同。
首先，让我们看一下示例3-1，以了解该管道的第一部分的缩写代码是什么样子的。
示例3-1。计算会话长度
1 | PCollection<Double> mobileSessions = IO.read(new MobileInputSource()) |
在这里，我们独立读取每个输入，而不是之前按团队键入我们的集合，在这个示例中，我们按用户键入。之后，对于每个管道的第一阶段，我们窗口进入会话，然后调用名为CalculateWindowLength的自定义PTransform。这个PTransform仅通过键（即用户）进行分组，然后通过将当前窗口的大小视为该窗口的值来计算每个用户会话长度。在这种情况下，我们对默认触发器（AtWatermark）和累积模式（discardingFiredPanes）设置感到满意，但是为了完整起见，我已经列出了它们。两个特定用户的每个管道的输出可能看起来像图3-4。
图3-4。两个不同输入管道中的每个用户会话长度
因为我们需要跨多个阶段跟踪数据，所以我们将与Mobile Scores相关的一切都记录为红色，与Console Scores相关的一切都记录为蓝色，而Figure 3-5中的平均会话长度的水位线和输出为黄色。
我们已经回答了计算单个会话长度的“什么”，“何地”，“何时”和“如何”这四个问题。接下来，我们将再次回答这些问题，将这些会话长度转换为固定时间窗口内的全局会话长度平均值。这需要我们首先将两个数据源展平为一个，然后重新窗口进入固定窗口；我们已经捕捉到会话中重要的本质，在我们计算的会话长度值中，我们现在想在一天中的一致时间窗口内计算这些会话的全局平均值。示例3-2显示了此代码。
示例3-2。计算会话长度
1 | PCollection<Double> mobileSessions = IO.read(new MobileInputSource()) |
如果我们看到这个管道在运行，它看起来会像图3-5。与以前一样，两个输入管道正在计算移动和控制台玩家的单个会话长度。然后，这些会话长度进入管道的第二个阶段，在其中计算了固定窗口内的全局会话长度平均值。
图3-5。移动和控制台游戏会话的平均会话长度
鉴于这里发生了很多事情，让我们走一遍这个例子。这里的两个重要点是：
- 每个Mobile Sessions和Console Sessions阶段的输出水位线至少与每个对应的输入水位线一样旧，并且实际上要旧一些。这是因为在真实系统中，计算答案需要时间，我们不允许输出水位线在给定输入的处理完成之前向前推进。
- 平均会话长度阶段的输入水位线是两个直接上游阶段的输出水位线的最小值。
结果是下游输入水位线是上游输出水位线的最小组合的别名。请注意，这与本章前面对这两种类型的水位线的定义相匹配。还要注意，下游的水位线进一步落后，捕捉到上游阶段将比后续阶段超前一些时间的直观概念。
这里值得注意的一个观察是，我们能够在示例3-1中再次询问这些问题，以从根本上改变管道的结果。以前，我们只是计算单个用户会话长度，现在我们计算了两分钟的全局会话长度平均值。这提供了对我们的游戏玩家的整体行为更深入的了解，让您微小地了解了简单数据转换和真正的数据科学之间的差异。
更好的是，既然我们了解了这个管道运作的基础知识，我们现在可以更仔细地查看与再次询问这四个问题相关的更微妙的问题：输出时间戳。
水位线传播和输出时间戳#
在图3-5中，我忽略了一些输出时间戳的细节。但是，如果您仔细观察图中的第二个阶段，您会发现从第一个阶段输出的每个数据都被赋予了一个时间戳，该时间戳与其窗口的结束时间相匹配。虽然这是一个相当自然的输出时间戳选择，但它并不是唯一有效的选择。正如您在本章早期所知道的，水位线永远不允许后移。在这种限制下，您可以推断出对于给定窗口的有效时间戳范围始于窗口中最早的非延迟记录的时间戳（因为只有非延迟记录保证保持水位线），并延伸到正无穷。这是相当多的选择。然而，在实践中，通常只有几个选择在大多数情况下是有意义的：
窗口结束时间
如果您希望输出时间戳代表窗口边界，则使用窗口结束时间是唯一安全的选择。正如我们很快就会看到的那样，这也允许水位线的平稳进展，比其他选项更加平稳。
第一个非延迟元素的时间戳
在您希望尽可能保守地保持水位线时，使用第一个非延迟元素的时间戳是一个不错的选择。然而，这种权衡的代价是水位线的进展可能会更加受到阻碍，我们很快也会看到。
特定元素的时间戳
对于某些用例，某些其他任意时间戳（从系统的角度来看）的时间戳是正确的选择。想象一下这样一个用例，您正在将查询流连接到该查询的结果流的点击流上。在执行连接之后，一些系统会发现查询的时间戳更有用；其他人则更喜欢单击的时间戳。只要它对应于没有延迟到达的元素，任何这样的时间戳都是从水位线正确性的角度来看是有效的。
考虑了一些备选项之后，让我们看看输出时间戳选择对整个管道的影响。为了使变化尽可能显著，在示例3-3和图3-6中，我们将使用最早的时间戳作为窗口的时间戳：第一个非延迟元素的时间戳。
示例3-3。平均会话长度管道，设置为最早元素的会话窗口输出时间戳
1 | PCollection<Double> mobileSessions = IO.read(new MobileInputSource())` |
图3-6。以最早元素的时间戳为输出时间戳的会话平均长度
为了突出输出时间戳选择的影响，请查看第一阶段中的虚线，显示每个阶段的输出水位线被保持的时间。由于我们选择了时间戳，输出水位线后移，与图3-7和3-8相比，在这些示例中，输出时间戳被选择为窗口结束时间。从这个图表中，您可以看出第二个阶段的输入水位线因此也被推迟了。
图3-7。具有不同窗口输出时间戳的水位线和结果的比较。
本图中的水位线对应于会话窗口的结束时间戳（即图3-5）。
图3-8。在这个图中，水位线位于会话窗口的开始处（即图3-6）。
我们可以看到，这个图中的水位线线被推迟了，导致平均会话长度不同。
就与图3-7相比这个版本的区别，值得注意的有两点：
水位线延迟
与图3-5相比，图3-6中的水位线进展要慢得多。这是因为第一个阶段的输出水位线被保留到每个窗口中的第一个元素的时间戳，直到该窗口的输入变得完整。只有在给定的窗口已经成为物化后，输出水位线（以及下游输入水位线）才被允许向前推进。
语义差异
因为现在会话时间戳被分配为与会话中最早的非延迟元素相匹配，所以当我们在下一个阶段计算会话长度平均值时，单个会话经常会落入不同的固定窗口桶中。两种选择都没有本质上的对错；它们只是不同的。但是，重要的是要理解它们将是不同的，并且具有它们将不同的直觉，以便在需要时为您特定的用例做出正确的选择。
重叠窗口的棘手情况#
关于输出时间戳的另一个微妙但重要的问题是如何处理滑动窗口。将输出时间戳设置为最早元素的幼稚方法很容易导致下游的延迟，因为水位线被（正确地）阻止。要了解原因，请考虑具有两个阶段的示例管道，每个阶段都使用相同类型的滑动窗口。假设每个元素都在三个连续的窗口中结束。随着输入水位线的推进，滑动窗口的期望语义如下：
- 第一个窗口在第一阶段完成并向下游发出。
- 然后在第二阶段中，第一个窗口完成并也可以向下游发出。
- 一段时间后，第二个窗口在第一阶段完成…等等。
然而，如果将输出时间戳选择为面板中第一个非延迟元素的时间戳，实际发生的是以下情况：
- 第一个窗口在第一阶段完成并向下游发出。
- 第二阶段的第一个窗口仍无法完成，因为其输入水位线被上游第二个和第三个窗口的输出水位线所阻止。这些水位线被正确地阻止，因为这些窗口的输出时间戳使用最早元素的时间戳。
- 第二个窗口在第一阶段完成并向下游发出。
- 在上游第三个窗口的带领下，第二阶段的第一个和第二个窗口仍无法完成。
- 第三个窗口在第一阶段完成并向下游发出。
- 第二阶段的第一个，第二个和第三个窗口现在都能够完成，最终一次性发出所有三个窗口。
尽管这种窗口处理的结果是正确的，但这导致了结果以不必要的延迟方式被实现。因此，Beam对于重叠窗口具有特殊逻辑，以确保窗口N+1的输出时间戳始终大于窗口N的结束时间戳。
百分位数水位线#
到目前为止，我们一直关注以活动信息在阶段中的最小事件时间为度量的水位线。跟踪最小值允许系统知道所有先前时间戳都已经被考虑。另一方面，我们可以考虑活动消息的事件时间分布，并利用它来创建更细粒度的触发条件。
我们可以选择考虑分布的任何百分位数，并且说我们保证处理了这个百分比的所有具有较早时间戳的事件。
这个方案的优点是什么？如果对于业务逻辑来说，“主要正确”已经足够，那么百分位数水位线提供了一种机制，可以使水位线比通过从水位线中剔除长尾的异常数据来跟踪最小事件时间更快、更平稳。图3-9展示了一个紧凑的事件时间分布，其中90百分位数水位线接近100百分位数。图3-10展示了一个离群点更远的情况，因此90百分位数水位线显着领先于100百分位数水位线。通过从水位线中剔除异常数据，百分位数水位线仍然可以跟踪分布的大部分，而不会被异常数据拖延。
图3-9. 看起来正常的水位线直方图
图3-10. 含有异常值的水位线直方图
图3-11展示了使用分位数水位线绘制两分钟固定窗口边界的示例。我们可以基于到达数据的时间戳的分位数，通过百分位数水位线绘制早期边界。
图3-11. 百分位数水位线变化的影响。随着百分位数的增加，窗口中包含更多的事件：但是，材料化窗口的处理时间延迟也会增加。
图3-11展示了33百分位数、66百分位数和100百分位数（全）水位线，跟踪数据分布中的相应时间戳百分位数。如预期的那样，这些允许绘制较早的边界，而不是跟踪完整的100百分位数水位线。请注意，33百分位数和66百分位数水位线都允许较早的窗口触发，但以更多的数据被标记为延迟为代价。例如，对于第一个窗口[12:00，12:02)，基于33百分位数水位线关闭的窗口只包括四个事件，并在12:06的处理时间材料化结果。如果我们使用66百分位数水位线，则相同的事件时间窗口将包括七个事件，并在12:07的处理时间材料化。使用100百分位数水位线包括所有十个事件，并推迟材料化结果，直到12:08的处理时间。因此，百分位数水位线提供了一种调整材料化结果的延迟和精度之间权衡的方法。
处理时间水位线#
到目前为止，我们一直在看水位线与流经我们系统的数据有关的情况。我们已经看到了查看水位线如何帮助我们识别最旧数据和实时数据之间的总延迟。然而，这还不足以区分旧数据和延迟的系统。换句话说，仅仅通过我们迄今为止定义的事件时间水位线来检查，我们无法区分正在快速处理一小时前的数据和被延迟一个小时的实时数据的系统。
为了区分这一点，我们需要更多的东西：处理时间水位线。我们已经看到，在流处理系统中有两个时间域：处理时间和事件时间。到目前为止，我们完全在事件时间域中定义了水位线，作为流经系统的数据时间戳的函数。这是一个事件时间水位线。我们现在将同样的模型应用于处理时间域，定义一个处理时间水位线。
我们的流处理系统不断执行操作，例如在管道的当前或上游阶段之前执行的操作之间移动消息、读取或写入消息到持久状态或触发基于水位线进度的延迟聚合。因此，正如数据元素“流”通过系统一样，处理这些元素涉及的操作的级联也“流”通过系统。
我们以完全相同的方式定义处理时间水位线，就像我们迄今为止定义的事件时间水位线一样，只不过我们使用最旧尚未完成的操作的处理时间戳，而不是使用最旧尚未完成的工作的事件时间戳。延迟处理时间水位线的例子可能是从一个阶段到另一个阶段的消息传递被阻塞、读取状态或外部数据的I/O调用被阻塞，或者在处理过程中发生的阻止处理完成的异常。
因此，处理时间水位线提供了一个单独于数据延迟的处理延迟概念。为了理解这个区别的价值，考虑图3-12中查看事件时间水位线延迟的情况。
我们看到数据延迟单调递增，但是没有足够的信息来区分系统卡住和数据卡住的情况。只有查看处理时间水位线，如图3-13所示，我们才能区分这些情况。
图3-12. 事件时间水位线增加。从这些信息中无法知道这是由于数据缓冲还是系统处理延迟引起的。
图3-13. 处理时间水位线也在增加。这表明系统处理被延迟了。
在第一种情况（图3-12）中，当我们检查处理时间水位线延迟时，我们发现它也在增加。这告诉我们，系统中的一个操作被卡住了，卡住也导致数据延迟落后。这可能发生在网络问题阻止管道阶段之间的消息传递或发生故障并正在进行重试的情况下。一般来说，增长的处理时间水位线表示一个问题，阻止了必要于系统功能的操作完成，通常需要用户或管理员干预来解决。
在第二种情况下，如图3-14所示，处理时间水位线延迟很小。这告诉我们，没有卡住的操作。事件时间水位线延迟仍在增加，这表明我们有一些缓冲状态正在等待排空。例如，如果我们正在等待窗口边界来通过及时触发器发出聚合，那么我们可能会缓冲一些状态，这对应于管道的正常操作，如图3-15所示。
图3-14. 事件时间水位线延迟增加，处理时间水位线稳定。这表明数据在系统中被缓冲并等待处理，而不是系统操作阻止了数据处理的迹象。
图3-15. 固定窗口的水位线延迟。随着元素为每个窗口缓冲，事件时间水位线延迟增加，随着每个窗口的聚合通过及时触发器发出，事件时间水位线延迟减少，而处理时间水位线仅跟踪系统级延迟（在健康的管道中仍保持相对稳定）。
因此，处理时间水位线是区分系统延迟和数据延迟的有用工具。除了可见性外，我们可以在系统实现级别使用处理时间水位线来执行任务，例如临时状态的垃圾回收（Reuven在第5章中对一个示例进行了更多的讨论）。
案例研究#
既然我们已经奠定了水位线应该如何行为的基础，现在是时候看一下一些真实的系统，以了解不同
实现水位线机制的方式。我们希望这些能够为现实世界中的水位线在延迟和正确性以及可扩展性和可用性之间的权衡提供一些启示。
案例研究：Google Cloud Dataflow中的水位线#
在流处理系统中实现水位线有许多可能的方法。在这里，我们简要介绍在Google Cloud Dataflow中的实现情况，这是一个完全托管的服务，用于执行Apache Beam管道。Dataflow包括用于定义数据处理工作流程的SDK，以及在Google Cloud Platform资源上运行这些工作流程的Cloud Platform托管服务。
数据流通过将每个数据处理图中的数据处理步骤条带（划分）到多个物理工作器中，将每个工作器的可用键空间分割为键范围，并将每个范围分配给一个工作器。每当遇到具有不同键的GroupByKey操作时，数据必须被洗牌到相应的键。
图3-16描述了具有GroupByKey的处理图的逻辑表示。
图3-16。 GroupByKey步骤从另一个DoFn消耗数据。这意味着第一步的键和第二步的键之间存在数据洗牌。
虽然将键范围物理分配给工作器的方式可能如图3-17所示。
图3-17。两个步骤的键范围分配（条带）在可用工作器上。
在水位线传播部分，我们讨论了为每个步骤维护水位线的多个子组件。Dataflow跟踪每个组件的每个范围的水位线。然后，水位线聚合涉及计算所有范围中每个水位线的最小值，以确保满足以下保证：
- 所有范围都必须报告水位线。如果没有范围的水位线，我们无法推进水位线，因为必须将不报告的范围视为未知。
- 确保水位线单调递增。因为可能存在延迟数据，所以如果更新水位线会导致水位线后移，我们不能更新水位线。
Google Cloud Dataflow通过一个集中的聚合代理进行聚合。我们可以为了效率而分片这个代理。从正确性的角度来看，水位线聚合器作为水位线的“真正的数据源”。
确保分布式水位线聚合的正确性带来了某些挑战。重要的是不要过早地推进水位线，因为过早地推进水位线会将准时数据转换为延迟数据。具体而言，当物理分配被激活到工作器时，工作器对附加到键范围的持久状态维护租约，确保只有一个工作器可以对一个键的持久状态进行修改。为了保证水位线的正确性，必须确保仅当工作器进程仍然在其持久状态上保持租约时，才会将来自工作器进程的每个水位线更新纳入聚合;因此，水位线更新协议必须考虑状态所有权租约验证。
案例研究：Apache Flink中的水位线#
Apache Flink是一个开源的流处理框架，用于分布式、高性能、始终可用和准确的数据流应用程序。可以使用Flink运行Beam程序。在这样做时，Beam依赖于Flink中实现的流处理概念，例如水位线。与实现集中式水位线聚合的Google Cloud Dataflow不同，Flink在带内执行水位线跟踪和聚合。
为了理解这是如何工作的，让我们看看Flink管道，如图3-18所示。
图3-18。带有两个源和带内传播事件时间水位线的Flink管道
在此管道中，数据在两个源处生成。这些源还都生成与数据流同步带内发送的水位线“检查点”。这意味着当来自时间戳“53”的源A的水位线检查点被发出时，它保证不会从源A发出时间戳在“53”之后的非延迟数据消息。下游的“keyBy”操作员消耗输入数据和水位线检查点。随着新的水位线检查点被消耗，下游操作员的水位线视图被推进，可以发出下游操作员的新水位线检查点。
与依赖于集中聚合的Cloud Dataflow方法不同，选择在数据流中带内发送水位线检查点会导致一些有趣的权衡。
以下是带内水位线的一些优点：
减少水位线传播延迟，非常低延迟的水位线
因为不需要水位线数据穿过多个跳跃并等待集中聚合，所以使用带内方法更容易实现非常低延迟。水位线聚合没有单点故障
集中的水位线聚合代理不可用将导致整个管道中的水位线延迟。使用带内方法，管道的一部分不可用也不能导致整个管道的水位线延迟。内在可扩展性
尽管Cloud Dataflow在实践中具有良好的可扩展性，但是使用集中式水位线聚合服务需要更多的复杂性，而使用带内水位线具有隐式的可扩展性。
以下是带外水位线聚合的一些优点：
唯一的“真相”来源
为了可调试性、监视和其他应用程序（例如基于管道进度限制输入），拥有一个能够提供水位线值的服务比将水位线隐式地放在流中更有优势，因为系统的每个组件都有自己的部分视图。源水位线创建
某些源水位线需要全局信息。例如，源可能处于暂时闲置状态、数据速率较低或需要有关源或其他系统组件的带外信息来生成水位线。这在中央服务中更容易实现。有关示例，请参见下面针对Google Cloud Pub/Sub的源水位线案例研究。
案例研究：Google Cloud Pub/Sub 的源水位线#
Google Cloud Pub/Sub 是一种完全托管的实时消息传递服务，允许您在独立应用程序之间发送和接收消息。在这里，我们讨论如何为通过 Cloud Pub/Sub 发送到管道的数据创建一个合理的启发式水位线。
首先，我们需要描述一下 Pub/Sub 的工作原理。消息发布在 Pub/Sub 主题 上。任何数量的 Pub/Sub 订阅 可以订阅特定主题。相同的消息被发送到订阅给定主题的所有订阅。客户端通过提供的 ID 从订阅中拉取消息，并确认接收到特定消息。客户端无法选择拉取哪些消息，尽管 Pub/Sub 尽力提供最旧的消息，但没有硬性保证。
为了构建一个启发式，我们对将数据发送到 Pub/Sub 的源进行了一些假设。具体来说，我们假设原始数据的时间戳是“良好的”，换句话说，我们期望在将数据发送到 Pub/Sub 之前，源数据出现有限数量的时间戳乱序。任何以时间戳超出允许的乱序范围发送的数据都将被视为延迟数据。在我们当前的实现中，此边界至少为 10 秒，这意味着在发送到 Pub/Sub 之前重新排序时间戳最多可达到 10 秒。我们称这个值为估计范围。另一个看待这个的方式是，当管道完全赶上输入时，水位线将落后于实时 10 秒，以允许源可能的重新排序。如果管道积压，所有积压（而不仅仅是 10 秒的范围）都用于估计水位线。
我们在 Pub/Sub 中面临的挑战是什么？因为 Pub/Sub 不保证顺序，我们必须具有某种额外的元数据以了解有关积压的足够信息。幸运的是，Pub/Sub 提供了一个“最旧的未确认发布时间戳”的积压度量。这与我们的消息的事件时间戳不同，因为 Pub/Sub 对通过它发送的应用程序级元数据是不可知的；相反，这是消息被 Pub/Sub 摄取的时间戳。
这个测量不同于事件时间水位线。实际上，它是 Pub/Sub 消息传递的处理时间水位线。Pub/Sub 发布时间戳不等于事件时间戳，在发送历史（过去）数据的情况下，它可能会远离。这些时间戳的排序也可能是不同的，因为我们允许有限数量的重新排序，如前面所述。
但是，我们可以将其用作积压的度量，以了解有关积压中存在的事件时间戳的足够信息，从而可以创建如下合理的水位线。
我们在包含输入消息的主题中创建两个订阅：基本订阅，管道将实际使用它来读取要处理的数据，以及跟踪订阅，仅用于元数据，以执行水位线估计。
在图3-19中查看我们的基本订阅，我们可以看到消息可能乱序到达。我们使用 Pub/Sub 发布时间戳“pt”和事件时间戳“et”标记每条消息。请注意，这两个时间域可能是无关的。
图3-19。Pub/Sub订阅上到达的处理时间和事件时间时间戳
基本订阅中的一些消息未被确认，形成了积压。这可能是因为它们尚未被传递，或者它们可能已经被传递但尚未被处理。请记住，从此订阅拉取是分布在多个碎片上的。因此，仅仅通过查看基本订阅，无法确定我们的水位线应该是什么。
跟踪订阅（如图3-20所示）用于有效地检查基本订阅的积压，并获取积压中事件时间戳的最小值。通过在跟踪订阅上保持很少或没有积压，我们可以检查基本订阅的最旧未确认消息之前的消息。
图3-20。接收相同消息的附加“跟踪”订阅和“基本”订阅
我们通过确保从此订阅中拉取是计算廉价的来保持在跟踪订阅上。相反，如果我们在跟踪订阅上落后足够多，我们将停止推进水位线。为此，我们确保至少满足以下条件之一：
- 跟踪订阅足够领先于基本订阅。足够领先意味着跟踪订阅领先至少估计范围。这确保考虑了估计范围内的任何有界重新排序。
- 跟踪订阅足够接近实时。换句话说，跟踪订阅上没有积压。
我们尽快确认跟踪订阅上的消息，之后我们会持久保存有关消息的发布和事件时间戳的元数据。我们将此元数据以稀疏直方图格式存储，以最小化使用空间和持久写入的大小。
最后，我们确保具有足够的数据来进行合理的水位线估计。我们从我们的跟踪订阅中读取的事件时间戳带宽有最新的发布时间戳，这些时间戳比基本订阅的最旧未确认时间戳新，或者估计范围的宽度。这确保我们考虑了积压中的所有事件时间戳，或者如果积压很小，则考虑了最近的估计范围，以进行水位线估计。
最后，水位线值计算为带宽中的最小事件时间。
这种方法的正确性在于，输入中在重新排序限制范围内的所有时间戳将由水位线考虑，不会出现延迟数据。然而，它可能会产生过度保守的水位线，一种“进展过慢”的方式，如第2章所述。因为我们在跟踪订阅上考虑了基本订阅的最旧未确认消息之前的所有消息，所以我们可以将已经确认的消息的事件时间戳包括在水位线估计中。
此外，还有一些启发式方法以确保进展。在密集、频繁到达数据的情况下，此方法效果很好。在数据稀疏或不频繁的情况下，可能没有足够的最近消息来构建合理的估计。如果我们在订阅上没有看到数据超过两分钟（并且没有积压），我们将将水位线推进到接近实时。这确保即使没有更多的消息，水位线和管道也会继续前进。
以上所有内容都确保只要源数据事件时间戳的重新排序在估计范围内，就不会有额外的延迟数据。
摘要#
到目前为止，我们已经探讨了如何利用消息的事件时间来为流处理系统提供强大的进度定义。我们看到这种进度的概念随后如何帮助我们回答处理事件时间发生的位置以及处理时间产生的结果的时间。具体来说，我们看到了如何在源头创建水位线，也就是将数据注入到管道中的数据摄取点，然后在整个管道中传播水位线，以保留允许回答“何时”和“何处”的关键保证。我们还探讨了更改输出窗口时间戳对水位线的影响。最后，我们探讨了在大规模构建水位线时考虑的一些实际系统问题。
现在我们已经对水位线在幕后的工作方式有了牢固的基础，我们可以深入研究它们在使用窗口和触发器回答更复杂的查询时可以为我们做些什么。请参阅第4章。
- 注意到了单调性的提及；我们尚未讨论如何实现这一点。实际上，到目前为止的讨论中没有提到单调性。如果我们仅考虑最旧的飞行事件时间，水位线不总是单调的，因为我们对输入没有做出任何假设。我们稍后回到这个讨论。
- 要精确，不仅需要日志数量是静态的，而且系统必须事先知道任何给定时间的日志数量。一个更复杂的输入源，由动态选择的输入日志组成，例如Pravega，也可以用于构建完美的水位线。只有在任何给定时间动态集合中存在的日志数量是未知的（如下一节中的示例）时，才必须回退到启发式水位线。
- 请注意，当我说“流经系统”时，并不一定意味着它们沿着正常数据的相同路径流动。它们可能（如Apache Flink中），但它们也可能以带外方式传输（如MillWheel / Cloud Dataflow）。
- 从水位线正确性的角度来看，“窗口”的开始不是一个安全的选择，因为窗口中的第一个元素通常在窗口本身的开始之后出现，这意味着水位线不保证被拖延到窗口的开始位置。
- 这里描述的分位数水位线触发方案目前尚未由Beam实现；然而，其他系统（如MillWheel）实现了这一点。
- 有关Flink水位线的更多信息，请参阅Flink文档。