flink基础知识----水印处理_duration.standardminutes

水印

到目前为止，我们一直在从管道作者或数据科学家的角度来看待流处理。第2章介绍了水印作为回答事件时间处理发生位置以及处理时间结果何时实现的基本问题的答案的一部分。在本章中，我们处理相同的问题，而不是从流处理系统的底层机制的角度来看。查看这些机制将有助于我们激发，理解和应用水印的概念。我们将讨论如何在数据入口处创建水印，它们如何在数据处理管道中传播，以及它们如何影响输出时间戳。我们还演示了水印如何保留必要的保证，以便在处理无界数据时回答处理事件时数据的位置以及何时实现数据的问题。

定义
考虑任何连续摄取数据和输出结果的管道。我们希望解决一个问题，即何时可以安全地调用事件时间窗口，这意味着窗口不再需要更多数据。为此，我们想要描述管道相对于其无界输入所取得的进展。
解决事件时间窗口问题的一种简单方法是简单地将事件时间窗口基于当前处理时间。正如我们在第1章中看到的那样，我们很快遇到了故障 - 数据处理和传输不是瞬时的，因此处理和事件时间几乎不相等。
我们管道中的任何打嗝或尖峰可能会导致我们错误地将消息分配给窗口。最终，这种策略失败了，因为我们没有强有力的方法来对这些窗口做出任何保证。
另一种直观但最终不正确的方法是考虑管道处理的消息速率。虽然这是一个有趣的指标，但是随着输入的变化，预期结果的可变性，可用于处理的资源等，速率可能会随意变化。更重要的是，费率无助于回答完整性的基本问题。具体来说，当我们看到特定时间间隔内的所有消息时，速率并没有告诉我们。在现实世界的系统中，将存在消息没有通过系统进展的情况。
这可能是暂时性错误（例如崩溃，网络故障，机器停机）或持久性错误的结果，例如需要更改应用程序逻辑或其他手动干预以解决的应用程序级故障。当然，如果发生大量故障，处理速率指标可能是检测此问题的良好代理。但是，速率指标永远不会告诉我们单个消息无法通过我们的管道取得进展。然而，即使是单个这样的消息也可以任意地影响输出结果的正确性。
我们需要更有力的衡量进展。为了到达那里，我们对流数据做了一个基本假设：每条消息都有一个关联的逻辑事件时间戳。在连续到达无界数据的情况下，这种假设是合理的，因为这意味着连续生成输入数据。在大多数情况下，我们可以将原始事件发生的时间作为其逻辑事件时间戳。使用包含事件时间戳的所有输入消息，我们可以检查任何管道中的此类时间戳的分布。这样的管道可以被分发以在许多代理上并行处理并消耗输入消息而不保证各个分片之间的排序。因此，此管道中活动的正在进行的消息的事件时间戳集将形成分布，如图3-1所示。

消息由管道摄取，处理并最终标记为已完成。每条消息都是“在飞行中”，意味着它已被接收但尚未完成或“已完成”，这意味着不再需要代表此消息进行处理。如果我们按事件时间检查消息的分布，它将如图3-1所示。
随着时间的推移，更多的消息将被添加到右侧的“飞行中”分发中，并且来自分发的“飞行中”部分的更多消息将被完成并移动到“完成”分发中。

图3-1。在流式传输管道中分配正在进行的和已完成的消息事件时间。新消息作为输入到达并保持“在飞行中”直到它们的处理完成。 “飞行中”分布的最左边缘对应于任何给定时刻最老的未处理元素。

此分布上有一个关键点，位于“飞行中”分布的最左边，对应于我们管道的任何未处理消息的最早事件时间戳。我们使用此值来定义水印：水印是尚未完成的最早工作的单调增加的时间戳。
00:00 / 00:00 1 此定义提供了两个基本属性，使其有用：
1.完整性 如果水印已超过某个时间戳T，我们可以通过其单调属性保证不会再进行处理在T处或之前的准时（非数据）事件,
因此，我们可以在T处或之前正确地发出任何聚合。换句话说，水印允许我们知道何时关闭窗口是正确的。

2.可见性 如果消息因任何原因卡在我们的管道中，则水印无法前进。此外，我们将通过检查阻止水印前进的消息来找到问题的根源。

源水印
创建这些水印来自哪里？要为数据源建立水印，我们必须为从该源进入管道的每条消息分配逻辑事件时间戳。正如第2章告诉我们的那样，所有水印创建都属于两大类：完美或启发式。为了提醒自己完美和启发式水印之间的区别，让我们看一下图3-2，它展示了第2章中的窗口求和示例。

图3-2。

完美（左）和启发（右）水印的窗口求和00:00 / 00:00

请注意，区别特征是 完美水印确保水印考虑所有数据，而启发式水印允许一些后期数据元素。
在水印创建为完美或启发后，水印在整个管道的其余部分保持不变。至于是什么使水印创作完美或启发式，它在很大程度上取决于消耗的源的性质。为了了解原因，让我们看一下每种水印创建的几个例子。

完美的水印
创建完美的水印创建为传入的消息分配时间戳，使得产生的水印严格保证不会再从该源再次看到事件时间小于水印的数据。

使用完美水印创建的管道永远不必处理后期数据;也就是说，在水印之后到达的数据已超过新到达消息的事件时间。然而，完美的水印创建需要完美的输入知识，因此对于许多真实的分布式输入源是不切实际的。以下是一些可以创建完美水印的用例示例：Ingress timestamping将进入时间指定为进入系统的数据的事件时间的源可以创建完美的水印。在这种情况下，源水印只是跟踪管道所观察到的当前处理时间。
这基本上是几乎所有支持2016年之前窗口化的流媒体系统使用的方法。
因为事件时间是从单个单调增加的源（实际处理时间）分配的，所以系统因此完全了解数据流中接下来会有哪些时间戳。结果，事件时间进度和窗口语义变得非常容易推理。当然，缺点是水印与数据本身的事件时间无关;这些事件时间被有效地丢弃，而水印只是跟踪数据相对于其到达系统的进度。

时间排序
日志的静态集合时间排序日志的静态大小输入源（例如，具有静态分区集的Apache Kafka主题，其中源的每个分区包含单调增加的事件时间）将是相对简单的源代码。创造一个完美的水印。为此，源将简单地跟踪已知和静态源分区集合上的未处理数据的最小事件2 时间（即，每个分区中最近读取的记录的事件时间的最小值）。
与前面提到的入口时间戳类似，由于已知跨越静态分区集的事件时间单调增加，因此系统完全了解下一个时间戳将会到来。这实际上是一种有界无序处理的形式;已知的一组分区中的无序量受这些分区中观察到的最小事件时间的限制。

通常，您可以保证在分区内单调增加时间戳的唯一方法是，在将数据写入其中时分配这些分区中的时间戳;例如，通过web前端将事件直接记录到Kafka中。尽管仍然是有限的用例，但这肯定比到达数据处理系统时的入口时间戳更有用，因为水印跟踪基础数据的有意义的事件时间。

启发式水印创建另一方面，启发式水印创建创建的水印仅仅是估计不会再次看到事件时间小于水印的数据的估计。使用启发式水印创建的管道可能需要处理一些后期数据。延迟数据是在水印超过此数据的事件时间之后到达的任何数据。只有启发式水印创建才能生成后期数据。如果启发式算法相当好，则后期数据的数量可能非常小，并且水印仍然可用作完成估计。如果要支持需要正确性的用例（例如，诸如计费之类的事情），系统仍然需要为用户提供处理延迟数据的方法。

对于许多现实世界的分布式输入源，构建完美水印在计算上或操作上都是不切实际的，但仍然可以通过利用输入数据源的结构特征来构建高度准确的启发式水印。以下是启发式水印（质量不同）的两个示例：动态时间排序日志集考虑一组动态结构化日志文件（每个单独文件包含相对于同一文件中其他记录单调增加事件时间的记录但是文件之间没有固定的事件时间关系），其中在运行时不知道完整的预期日志文件集（即Kafka用语中的分区）。这些投入通常存在于由许多独立团队构建和管理的全球规模服务中。在这种用例中，在输入上创建完美的水印是难以处理的，但是创建精确的启发式水印是非常有可能的。

通过跟踪现有日志文件集中未处理数据的最小事件时间，监控增长率，并利用网络拓扑和带宽可用性等外部信息，您可以创建非常准确的水印，即使缺乏对所有日期文件的完美了解投入。这种类型的输入源是Google发现的常见类型的无界数据集之一，因此我们在为这些场景创建和分析水印质量方面拥有丰富的经验，并且已经看到它们在许多用例中都具有良好的效果。
Google Cloud Pub / Sub Cloud Pub / Sub是一个有趣的用例。 Pub / Sub目前不保证按顺序交付;即使单个发布者按顺序发布两条消息，也有可能（通常很小）它们可能无序传送（这是由于底层架构的动态特性，允许透明扩展到非常高的级别没有用户干预的吞吐量）。因此，无法保证Cloud Pub / Sub的完美水印。但是，Cloud Dataflow团队通过利用Cloud Pub / Sub中有关数据的可用知识，构建了一个相当准确的启发式水印。本章稍后将详细讨论该启发式的实现。

考虑用户玩移动游戏的示例，并将他们的分数发送到我们的管道进行处理：您通常可以假设对于使用移动设备进行输入的任何源，通常不可能提供完美的水印。由于设备长时间脱机的问题，没有办法对这种数据源提供任何合理的绝对完整性估计。但是，您可以设想构建一个水印，准确跟踪当前在线设备的输入完整性，类似于刚才描述的Google Pub / Sub水印。从提供低延迟结果的角度来看，积极在线的用户可能是最相关的用户子集，因此这通常不像您最初想象的那样缺点。

从广义上讲，启发式水印创建越多，对源的了解越多，启发式越好，后期数据项就越少。鉴于源的类型，事件的分布和使用模式会有很大差异，因此没有一个通用的解决方案。但是在任何一种情况（完美或启发式）中，在输入源处创建水印之后，系统可以完美地通过管道传播水印。这意味着完美的水印将保持完美的下游，启发式水印将严格保持与启动时一样的启发式。

这是水印方法的好处：您可以将管道中完整性跟踪的复杂性完全降低到在源头创建水印的问题。

水印传播到目前为止，我们只考虑了单个操作或阶段环境中输入的水印。但是，大多数真实世界的管道都包含多个阶段。了解水印如何在独立阶段传播对于了解它们如何影响整个管道以及观察到的结果延迟非常重要。
管道阶段每次管道将数据组合在一起时，通常需要不同的阶段。例如，如果您有一个使用原始数据的管道，计算了一些每用户聚合，然后使用这些每用户聚合来计算一些每个团队的聚合，那么您最终可能会得到一个三阶段管道：一个消耗原始的，未分组的数据按用户分组数据和计算每用户聚合数按团队分组数据和计算每个团队的聚合数我们将在第6章中详细了解分组对管道形状的影响。

水印在输入源处创建，如上一节中所述。然后，随着数据通过它，它们在概念上流过系统。您可以以不同的粒度级别跟踪水印。对于包含多个不同阶段的管道，每个阶段可能跟踪其自己的水印，其值是其前面的所有输入和阶段的函数。因此，在管道中稍后出现的阶段将具有过去进一步的水印（因为他们已经看到较少的总体输入）。
我们可以在管道中的任何单个操作或阶段的边界定义水印。这不仅有助于了解管道中每个阶段的相对进展，而且可以为每个阶段独立和尽快地及时发布结果。我们对阶段边界处的水印给出以下定义：输入水印，其捕获该阶段上游的所有进度（即，该阶段的输入有多完整）。
对于源，输入水印是源特定功能，为输入数据创建水印。对于非源阶段，输入水印被定义为其所有上游源和阶段的所有分片/分区/实例的输出水印的最小值。
输出水印，捕获舞台本身的进度，基本上定义为舞台输入水印的最小值和舞台内所有非相关数据活动消息的事件时间。 “活动”所包含的内容在某种程度上取决于给定阶段实际执行的操作以及流处理系统的实现。它通常包括缓冲用于聚合但尚未实现下游的数据，等待传输到下游阶段的输出数据，等等。
为特定阶段定义输入和输出水印的一个很好的特性是我们可以使用它们来计算阶段引入的事件 - 时间延迟量。从输入水印的值中减去阶段输出水印的值，得出阶段引入的事件 - 时间延迟或滞后量。这种滞后是每个阶段的输出将在实际时间之后延迟多远的概念。作为示例，执行10秒窗口聚合的阶段将具有10秒或更长的滞后，这意味着阶段的输出将至少在输入和实时之后延迟很多。输入和输出水印的定义在整个管道中提供水印的递归关系。管道中的每个后续阶段根据阶段的事件时间滞后将水印延迟为必要的。

每个阶段的处理也不是单一的。我们可以将一个阶段内的处理分成具有多个概念组件的流程，每个概念组件都有助于输出水印。如前所述，这些组件的确切性质取决于阶段执行的操作和系统的实现。从概念上讲，每个这样的组件都充当缓冲区，其中活动消息可以驻留，直到某些操作完成。例如，当数据到达时，它被缓冲以进行处理。然后，处理可以将数据写入状态以供稍后延迟聚合。触发时，延迟聚合可能会将结果写入等待下游阶段消耗的输出缓冲区，如图3-3所示。

图3-3。流系统级的示例系统组件，包含飞行中数据的缓冲区。每个都将具有相关的水印跟踪，并且该级的总输出水印将是所有这些缓冲器中的水印的最小值。

我们可以用自己的水印跟踪每个这样的缓冲区。每个级的缓冲区中的最小水印形成了级的输出水印。因此，输出水印可以是以下的最小值：每个发送阶段的每源水印。
每个外部输入水印 - 用于管道外部的源每个状态组件水印 - 适用于每种类型的状态，可以为每个接收阶段提供每输出缓冲水印在此粒度级别提供水印也可提供更好的可见性进入系统的行为。水印跟踪系统中各种缓冲区的消息位置，以便更容易地诊断卡住。

了解水印传播为了更好地理解输入和输出水印之间的关系以及它们如何影响水印传播，让我们看一个例子。我们考虑游戏分数，但不是计算团队分数的总和，我们将尝试衡量用户参与度。

我们将首先计算每用户会话长度，假设用户与游戏保持联系的时间量是他们享受它的程度的合理代理。在回答了我们的四个问题以计算会话长度后，我们将再次回答它们以计算固定时间段内的平均会话长度。
为了使我们的示例更有趣，我们假设我们正在使用两个数据集，一个用于移动分数，另一个用于控制台分数。我们希望通过这两个独立数据集上的整数求和来执行相同的分数计算。一个管道正在计算在移动设备上玩的用户的分数，而另一个管道用于在家庭游戏控制台上玩的用户，可能是由于针对不同平台采用的不同数据收集策略。重要的是，这两个阶段在不同的数据上执行相同的操作，因此具有非常不同的输出水印。
首先，让我们看一下例3-1，看看这个管道的第一部分的缩写代码是什么样的。

Example 3-1. Calculating session lengths
PCollection mobileSessions = IO.read(new MobileInputSource())
.apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1)))
.triggering(AtWatermark())
.discardingFiredPanes())
.apply(CalculateWindowLength());
PCollection consoleSessions = IO.read(new ConsoleInputSource())
.apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1)))
.triggering(AtWatermark())
.discardingFiredPanes())
.apply(CalculateWindowLength());

在这里，我们独立地阅读每个输入，而之前我们按团队键入我们的集合，在这个例子中我们按用户键入。
之后，对于每个管道的第一个阶段，我们进入会话窗口，然后调用名为CalculateWindowLength的自定义PTransform。 该PTransform只是按键分组（即用户），然后通过将当前窗口的大小视为该窗口的值来计算每个用户的会话长度。 在这种情况下，我们可以使用默认触发器（AtWatermark）和累积模式（discardingFiredPanes）设置，但我已明确列出它们的完整性。 两个特定用户的每个管道的输出可能如图3-4所示。

因为我们需要跨多个阶段跟踪数据，所以我们跟踪与红色移动分数相关的所有内容，与蓝色控制台分数相关的所有内容，而图3-5中的平均会话长度的水印和输出为黄色。
我们已经回答了计算单个会话长度的内容，位置，时间和方式这四个问题。 接下来，我们将第二次回答它们，将这些会话长度转换为固定时间窗口内的全局会话长度平均值。 这要求我们首先将我们的两个数据源拼合成一个，然后重新进入固定窗口; 我们已经在我们计算的会话长度值中捕获了会话的重要本质，现在我们想要在一天中的一致时间窗口内计算这些会话的全局平均值。 例3-2显示了这个代码。

Example 3-2. Calculating session lengths
PCollection mobileSessions = IO.read(new MobileInputSource())
.apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1)))
.triggering(AtWatermark())
.discardingFiredPanes())
.apply(CalculateWindowLength());
PCollection consoleSessions = IO.read(new ConsoleInputSource())
.apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1)))
.triggering(AtWatermark())
.discardingFiredPanes())
.apply(CalculateWindowLength());
PCollection averageSessionLengths = PCollectionList
.of(mobileSessions).and(consoleSessions)
.apply(Flatten.pCollections())
.apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
.triggering(AtWatermark())
.apply(Mean.globally());

如果我们看到这个管道正在运行，它将如图3-5所示。和以前一样，两个输入管道正在计算移动和控制台播放器的各个会话长度。然后，这些会话长度将进入管道的第二阶段，其中全局会话长度平均值在固定窗口中计算。
图3-5。移动和控制台游戏会话的平均会话长度00:00 / 00:00 122让我们来看看这个例子中的一些，因为有很多事情要发生。
这里的两个要点是：每个移动会话和控制台会话阶段的输出水印至少与每个相应的输入水印一样长，实际上有点旧。这是因为在实际系统中，计算答案需要花费时间，并且我们不允许输出水印前进，直到给定输入的处理完成为止。
Average Session Lengths阶段的输入水印是直接上游两个阶段的输出水印的最小值。
结果是下游输入水印是上游输出水印的最小组成的别名。请注意，这与本章前面这两种类型的水印的定义相匹配。还要注意过去下游的水印是如何进一步的，捕捉直观的概念，即上游阶段将比其后续阶段更进一步提前。
这里值得一提的是，我们能够如何干净地在例3-1中再次提出问题，以大幅改变管道的结果。在我们简单地计算每用户会话长度之前，我们现在计算两分钟的全局会话长度平均值。这样可以更加深入地了解玩游戏的用户的整体行为，并简要介绍简单数据转换与真实数据科学之间的差异。
更好的是，既然我们已经理解了这个管道如何运作的基础知识，我们可以更仔细地看一下与再次提出四个问题相关的一个更微妙的问题：输出时间戳。

水印传播和输出时间戳在图3-5中，我隐藏了输出时间戳的一些细节。但是如果仔细观察图中的第二阶段，可以看到第一阶段的每个输出都被分配了一个与其窗口末端相匹配的时间戳。虽然这是输出时间戳的一个相当自然的选择，但它不是唯一有效的选择。正如您在本章前面所知，水印永远不会被允许向后移动。鉴于该限制，您可以推断出给定窗口的有效时间戳范围以窗口中最早的非最大记录的时间戳开始（因为只保证不存在水印的非记录记录）并一直延伸到正无穷大。这是很多选择。
然而，在实践中，在大多数情况下，往往只有少数选择是有意义的：窗口结束如果希望输出时间戳代表窗口边界，则使用窗口末尾是唯一安全的选择。正如我们稍后将看到的那样，它还允许所有选项中最平滑的水印进展。
第一个nonlate元素的时间戳当你想让你的水印尽可能保守时，使用第一个nonlate元素的时间戳是个不错的选择。然而，权衡是水印进展可能会更加受阻，我们也会很快看到。
特定元素的时间戳对于某些用例，某些其他任意（从系统的角度来看）元素的时间戳是正确的选择。想象一个用例，在这个用例中，您将查询流加入到该查询的结果点击流中。执行连接后，某些系统会发现查询的时间戳更有用;其他人会更喜欢点击的时间戳。从水印4 124正确性的角度来看，任何这样的时间戳都是有效的，只要它对应于没有迟到的元素。

在考虑了输出时间戳的一些备用选项之后，让我们看一下输出时间戳的选择对整个管道的影响。为了使更改尽可能引人注目，在例3- 3和图3-6中，我们将切换到使用窗口可能的最早时间戳：第一个nonlate元素的时间戳作为窗口的时间戳。
例3-3。平均会话长度管道，输出最早元素设置的会话窗口的时间戳

PCollection mobileSessions = IO.read(new MobileInputSource())
.apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1)))
.triggering(AtWatermark())
.withTimestampCombiner(EARLIEST)
.discardingFiredPanes())
.apply(CalculateWindowLength());
PCollection consoleSessions = IO.read(new ConsoleInputSource())
.apply(Window.into(Sessions.withGapDuration(Duration.standardMinutes(1)))
.triggering(AtWatermark())
.withTimestampCombiner(EARLIEST)
.discardingFiredPanes())
.apply(CalculateWindowLength());
PCollection averageSessionLengths = PCollectionList
.of(mobileSessions).and(consoleSessions)
.apply(Flatten.pCollections())
.apply(Window.into(FixedWindows.of(Duration.standardMinutes(2)))
.triggering(AtWatermark())
.apply(Mean.globally());

为了帮助调出输出时间戳选择的效果，请查看第一阶段中的虚线，以显示每个阶段的输出水印所处的位置。 与图3-7和3-8相比，输出水印被我们选择的时间戳延迟，其中输出时间戳被选择为窗口的结束。 从该图中可以看出，第二阶段的输入水印随后也被延迟。

为了帮助调出输出时间戳选择的效果，请查看第一阶段中的虚线，以显示每个阶段的输出水印所处的位置。

与图3-7和3-8相比，

输出水印被我们选择的时间戳延迟，其中输出时间戳被选择为窗口的结束。从该图中可以看出，第二阶段的输入水印随后也被延迟。
重叠Windows的棘手案例关于输出时间戳的另一个微妙但重要的问题是如何处理滑动窗口。将输出时间戳设置为最早元素的简单方法很容易导致下游延迟，因为水印被（正确地）阻止。要了解原因，请考虑具有两个阶段的示例管道，每个阶段使用相同类型的滑动窗口。假设每个元素最终连续三个窗口。
随着输入水印的进展，在这种情况下滑动窗口的所需语义如下：第一窗口在第一阶段完成并向下游发射。
然后第一个窗口在第二个阶段完成，也可以向下游发射。
一段时间后，第二个窗口在第一阶段完成…依此类推。
但是，如果选择输出时间戳作为窗格中第一个nonlate元素的时间戳，则实际发生的情况如下：第一个窗口在第一个阶段完成并向下游发出。
第二阶段中的第一个窗口仍然无法完成，因为其输入水印被上游第二和第三窗口的输出水印所阻挡。这些水印正在被阻止，因为最早的元素时间戳被用作那些窗口的输出时间戳。
第二个窗口在第一个阶段完成并向下游发射。
第二阶段中的第一和第二窗口仍然无法完成，由上游的第三窗口保持。

第三个窗口在第一个阶段完成，并向下游发射。
第二阶段的第一，第二和第三窗口现在都能够完成，最终一举发射全部三个。
虽然这种窗口的结果是正确的，但这导致结果以不必要的延迟方式实现。因此，Beam具有用于重叠窗口的特殊逻辑，其确保窗口N 1的输出时间戳总是大于窗口N的结束。

Percentile Watermarks

到目前为止，我们一直关注水印，这是通过阶段中活动消息的最小事件时间来衡量的。跟踪最小值允许系统知道何时考虑了所有较早的时间戳。另一方面，我们可以考虑活动消息的事件时间戳的整个分布，并利用它来创建更细粒度的触发条件。
我们可以采用分布的任何百分位，而不是考虑分布的最小点，并且说我们保证已经使用更早的时间戳处理了所有事件的这个百分比。
这个方案有什么好处？如果对于业务逻辑“大多数”正确就足够了，百分位水印提供了一种机制，通过该机制，水印可以比通过从水印中丢弃分布的长尾中的异常值来跟踪最小事件时间更快更平稳地前进。图3-9显示了事件时间的紧凑分布，其中90百分位水印接近100百分位数。图3-10显示了异常值进一步落后的情况，因此90百分位水印明显领先于100百分位数。通过从水印中丢弃异常数据，百分位水印仍然可以跟踪分布的大部分而不会被异常值延迟。

图3-11显示了用于绘制两分钟固定窗口的窗口边界的百分位水印的示例。
我们可以根据百分位水印跟踪的到达数据的时间戳百分比来绘制早期边界。

图3-11显示了33百分位数，66百分位数和100百分位数（完整）水印，跟踪数据分布中各自的时间戳百分位数。正如预期的那样，这些允许比跟踪完整的100百分位水印更早地绘制边界。请注意，33和66百分位水印每个都允许更早触发窗口，但需要权衡标记更多数据的延迟。例如，对于第一个窗口[12:00,12：02]，基于33百分位水印关闭的窗口将仅包括四个事件并且在12:06处理时间实现结果。如果我们使用66百分位水印，则相同的事件时间窗口将包括七个事件，并在12:07处理时间实现。使用100百分位水印包括所有十个事件和延迟实现结果直到12:08处理时间。因此，在第134个水印的百分位数00:00 / 00:00提供了一种方法来调整具体化结果的延迟和结果的精确度之间的权衡。

处理时间水印

到目前为止，我们一直在研究水印，因为它们与流经我们系统的数据有关。我们已经看到了如何查看水印可以帮助我们确定最早的数据和实时之间的整体延迟。
但是，这还不足以区分旧数据和延迟系统。换句话说，通过仅检查我们迄今为止定义的事件时间水印，我们无法区分一小时前快速且无延迟地处理数据的系统，以及试图处理实际的系统时间数据并且已经延迟了一个小时。
为了做出这种区分，我们需要更多的东西：处理时间水印。我们已经看到流媒体系统中有两个时域：处理时间和事件时间。到目前为止，我们已经完全在事件时间域中定义了水印，作为流经系统的数据的时间戳的函数。这是一个事件时间水印。我们现在将相同的模型应用于处理时域以定义处理时间水印。
我们的流处理系统不断执行诸如在阶段之间改组消息，将消息读取或写入持久状态，或基于水印进度触发延迟聚合等操作。所有这些操作都是响应于在流水线的当前或上游阶段完成的先前操作而执行的。因此，正如数据元素“流过”系统一样，处理这些元素所涉及的一系列操作也“流动”通过系统。
我们以与定义事件时间水印完全相同的方式定义处理时间水印，除了使用尚未完成的最旧工作的事件时间时间戳，我们使用最旧操作的处理时间时间戳而不是尚未完成。处理时间水印的延迟的示例可以是从一个阶段到另一个阶段的卡住消息传递，对读取状态或外部数据的卡住I / O调用，或者在处理时阻止处理完成的异常。
因此，处理时间水印提供了与数据延迟分开的处理延迟的概念。要了解这种区别的价值，请考虑图3-12中的图表，其中我们查看事件时间水印延迟。
我们看到数据延迟是单调增加的，但是没有足够的信息来区分卡住系统和卡住数据的情况。只有通过查看处理时水印，如图3-13所示，我们才能区分这些情况。

在第一种情况下（图3-12），当我们检查处理时间水印延迟时，我们发现它也在增加。这告诉我们系统中的操作被卡住了，卡住也导致数据延迟落后。可能发生这种情况的一些现实示例是当网络问题阻止管道的各级之间的消息传递或者如果已经发生故障并且正在重试时。

通常，不断增长的处理时间水印指示阻止操作完成系统功能所必需的问题，并且通常涉及用户或管理员干预以解决。
在第二种情况下，如图3-14所示，处理时间水印延迟很小。这告诉我们没有卡住操作。事件时间水印延迟仍在增加，这表明我们有一些缓冲状态，我们正在等待消耗。例如，如果我们在等待窗口边界发出聚合时缓冲某个状态，并且对应于管道的正常操作，则这是可能的，如图3-15所示。

图3-15。 固定窗口的水印延迟。 事件时间水印延迟随着每个窗口的元素缓冲而增加，并且随着每个窗口的聚合通过接通时间触发发射而减少，而处理时间水印仅仅跟踪系统级延迟（在一个时间内保持相对稳定） 健康的管道）。
因此，处理时水印是区分系统延迟和数据延迟的有用工具。 除了可见性之外，我们还可以在系统实现级别使用处理时水印来执行诸如临时状态的垃圾收集之类的任务（Reuven在第5章中更多地讨论了这个例子）。