Chapter 2. The What, Where, When, and How of Data Processing（数据处理的内容，地点，时间和方式）

好的聚会人员，是时候具体了！

第1章主要关注三个方面：术语，准确定义当我使用“流”之类的重载术语时的含义；批处理与流式处理，比较两种类型的系统的理论功能，并假设使流式系统超出批处理对等系统仅需要两件事：正确性和时间推理工具；和数据处理模式，研究批处理和流处理系统在处理有界和无界数据时采用的概念方法。

在本章中，我们现在将进一步集中于第1章中的数据处理模式，但将在具体示例的上下文中更详细地介绍。待完成时，我们将介绍我认为是健壮的乱序数据处理所需的核心原则和概念。这些是进行时间推理的工具，可让您真正超越传统的批处理。

为了让您了解实际情况，我使用Apache Beam代码片段以及延时图来提供概念的直观表示。Apache Beam是用于批处理和流处理的统一编程模型和可移植性层，其中包含一组使用各种语言（例如Java和Python）的具体SDK。然后，可以在任何受支持的执行引擎（Apache Apex，Apache Flink，Apache Spark，Cloud Dataflow等）上移植运行用Apache Beam编写的管道。

我在这里使用Apache Beam作为示例，并不是因为这是一本Beam书（不是），而是因为它最完整地体现了本书中描述的概念。回到最初编写“ Streaming 102”时（回到当时仍是来自Google Cloud Dataflow的数据流模型而不是来自Apache Beam的Beam模型），它实际上是存在的唯一提供所有内容所需表达能力的系统。我们将在此处介绍的示例。一年半后，我很高兴地说，情况已经发生了很大变化，并且那里的大多数主要系统已经或正在朝着支持一种看起来很像本书中所描述的模型的方向发展。因此，请放心，尽管本文通过Beamlens进行了介绍，但我们在此介绍的概念将同样适用于您将遇到的大多数其他系统。

路线图

为了帮助奠定本章的基础，我想列出五个主要概念，这些概念将作为本章所有其余部分的基础，实际上，对于第一部分的其余大部分，我们已经介绍了其中的两个。

在第一章中，我首先确定了事件时间（事件发生的时间）和处理时间（在处理过程中观察到的时间）之间的关键区别。这为本书提出的主要论断之一奠定了基础：如果您既关注事件发生的正确性和上下文，则必须分析与事件固有时间相关的数据，而不是事件发生的处理时间。分析本身遇到的问题。

然后，我介绍了窗口化的概念（即沿时间边界划分数据集），这是一种常见的方法，用于应对技术上无限的数据源可能永远不会结束的事实。窗口策略的一些简单示例是固定窗口和滑动窗口，但是窗口类型更为复杂，例如会话（其中，窗口由数据本身的特征来定义；例如，捕获每个用户的活动会话，然后留一个空白）闲置状态）也得到了广泛的应用。

除了这两个概念，我们现在将进一步研究另外三个：

触发器

触发器是一种机制，用于声明何时应相对于某些外部信号实现窗口的输出。触发器提供了选择何时发出输出的灵活性。从某种意义上讲，您可以将它们视为用于指示何时应实现结果的流程控制机制。另一种看待它的方式是，触发器就像照相机上的快门释放按钮一样，使您可以在计算结果时声明何时拍摄快照。

触发器还使观察窗口演变时可以多次观察其输出。反过来，这为随着时间的流逝细化结果打开了大门，这允许在数据到达时提供推测性结果，以及处理上游数据（修订）随时间的变化或数据到达较晚的情况（例如，移动场景），某人离线时，他的电话会记录各种动作及其事件时间，然后在恢复连接后继续上传这些事件以进行处理）。

水印

水印是关于事件时间的输入完整性的概念。时间值为X的水印说明：“已观察到事件时间小于X的所有输入数据。” 这样，在观察无端数据源时，水印可作为进度的度量标准。在本章中，我们讨论了水印的基础知识，然后在第3章中，Slava深入探讨了水印。

累积

累积模式指定针对同一窗口观察到的多个结果之间的关系。这些结果可能会完全脱节。也就是说，代表随时间变化的独立增量，或者它们之间可能存在重叠。不同的累积模式具有不同的语义和与此相关的成本，因此可以在各种用例中找到适用性。

另外，因为我认为这样可以更轻松地理解所有这些概念之间的关系，所以我们在回答四个问题的结构内重新探讨了旧的概念，并探索了新的概念，我提出的所有问题对于每个无限制的数据处理问题都是至关重要的：

• 计算出什么结果？ 管道中转换的类型回答了这个问题。其中包括计算总和，构建直方图，训练机器学习模型等。从本质上讲，这也是经典批处理所回答的问题
• 结果在哪里计算？ 通过在管道中使用事件时间窗口可以回答此问题。这包括第1章中的窗口化的常见示例（固定，滑动和会话）。似乎没有开窗概念的用例（例如，与时间无关的处理;经典的批处理通常也属于此类）；以及其他更复杂的窗口类型，例如限时拍卖。还请注意，如果您将记录的到达时间指定为事件到达系统的时间，则它也可以包括处理时间窗口。
• 在处理时间何时实现？ 通过使用触发器和（可选）水印可以回答此问题。这个主题有无限的变化，但是最常见的模式是那些涉及重复更新的模式（即，物化视图语义），那些利用水印仅在认为相应的输入完成后才使用每个窗口提供单个输出的模式（即，基于每个窗口应用的经典批处理语义）或两者的某种组合。
• 细化结果如何关联？ 这个问题可以通过使用的累加类型来回答：丢弃（结果都是独立且不同的），累加（以后的结果以先前的结果为基础）或累加和收回（其中的累加值加上对结果的收回） 发出先前触发的值）。

在本书的其余部分中，我们将更详细地研究每个问题。而且，是的，我将把这种配色方案付诸实践，以期弄清楚在“什么/在哪里/何时/如何成语”中哪些概念与哪个问题有关。不客气。

批处理基础：什么和在哪里

好的，让我们开始这个聚会。第一站：批处理。

什么：转换

经典批处理中应用的转换回答了以下问题：“计算了什么结果？” 即使您可能已经熟悉经典的批处理，我们还是要从那里开始，因为它是我们添加所有其他概念的基础。

在本章的其余部分（以及实际上，贯穿本书的大部分内容），我们将看一个例子：在由9个值组成的简单数据集上计算带键整数和。假设我们已经编写了一个基于团队的手机游戏，并且想要建立一个通过汇总用户手机报告的个人得分来计算团队得分的管道。如果我们要在名为“ UserScores”的SQL表中捕获9个示例分数，则可能看起来像这样：

SELECT * FROM UserScores ORDER BY EventTime;

请注意，此示例中的所有分数均来自同一团队的用户；鉴于下面的图表中的维数有限，这是为了简化示例。而且由于我们是按团队分组，因此我们实际上只关心最后三列：

得分

与此事件相关的个人用户得分

活动时间

得分的活动时间；即得分发生的时间

ProcTime

分数的处理；也就是说，管道观察分数的时间

对于每个示例管道，我们将查看一个延时图表，突出显示数据随时间的变化情况。这些图在我们关注的时间的两个维度上绘制了我们的九个分数：x轴上的事件时间，y轴上的处理时间。图2-1说明了输入数据的静态图。

图2-1 九个输入记录，同时记录了事件时间和处理时间

随后的延时图可以是动画（Safari）或帧序列（打印和所有其他数字格式），使您可以看到随着时间的流逝如何处理数据（在第一次延时之后不久，将对此进行更多介绍）。图）。

每个示例之前都是Apache Beam Java SDK伪代码的简短摘要，以使管道的定义更加具体。从某种意义上来说，这是伪代码，我有时会弯腰规则以使示例更清楚，省略细节（例如使用具体的I / O源）或简化名称（Beam Java 2.x和更早版本中的触发器名称非常冗长） ；为清楚起见，我使用更简单的名称）。除了这些小事情，它是真实世界的Beam代码（本章中所有示例的GitHub上都提供了真实代码）。

如果您已经熟悉Spark或Flink之类的知识，则应该相对容易地了解Beam代码的功能。但是，为了让您对事物有所了解，Beam中有两个基本原语：

P系列

这些代表可以执行并行转换的数据集（可能是庞大的数据集）（因此名称的开头为“ P”）。

P转换

这些将应用于PCollection以创建新的PCollection。PTransform可以执行逐元素的转换，它们可以将多个元素分组/聚合在一起，也可以是其他PTrans形式的组合，如图2-2所示。

图2-2 转换类型
就我们的示例而言，我们通常假设我们从预加载的PCollection >开始，命名为“ input”（即，由Teams和Integer的键/值对组成的PCollection，其中 团队就像代表团队名称的字符串一样，而整数则是相应团队中任何个人的分数）。在现实世界的管道中，我们将通过从I / O源中读取原始数据（例如日志记录）的PCollection 并将其转换为PCollection >来获取输入。通过将日志记录解析为适当的键/值对。为了清楚起见，在第一个示例中，我包括了所有这些步骤的伪代码，但在后续示例中，我省略了I / O和解析。

因此，对于仅从I / O源中读取数据，解析团队/得分对并计算每组分数总和的管道，我们将具有示例2-1中所示的内容。

示例2-1 求和管道

PCollection<String> raw = IO.read(...);
PCollection<KV<Team, Integer>> input = raw.apply(new ParseFn());
PCollection<KV<Team, Integer>> totals = input.apply(Sum.integersPerKey());

从I / O源读取键/值数据，其中以Team（例如，团队名称的字符串）作为键，并使用Integer（例如，单个团队成员的分数）作为值。然后将每个关键点的值相加在一起，以在输出集合中生成每个关键点的总和（例如，团队总得分）。

对于接下来的所有示例，在看到描述我们正在分析的管道的代码片段之后，我们将查看一个延时图，该图显示了在单个键的具体数据集上该管道的执行情况。在真实的管道中，您可以想象类似的操作将在多台计算机上并行发生，但是就我们的示例而言，使事情变得简单起来会更加清楚。

如前所述，Safari版本以动画电影形式显示完整的执行过程，而印刷品和所有其他数字格式均使用静态的关键帧序列，以提供管道随着时间的流逝如何发展的感觉。在这两种情况下，我们还在www.streamingbook.net上提供了指向完整动画版本的URL。

每个图在两个维度上绘制输入和输出：事件时间（在x轴上）和处理时间（在y轴上）。因此，如通过管线观察到的实时从底部到顶部进行，如随着时间的进行在处理时间轴上上升的粗水平黑线所示。输入是圆圈，圆圈内的数字表示该特定记录的值。它们以浅灰色开始，然后随着管道的观察而变暗。

当管道观察值时，它会在中间状态下累积它们，并最终将汇总结果具体化为输出。状态和输出由矩形表示（状态为灰色，输出为蓝色），聚合值在顶部附近，矩形所覆盖的区域表示事件时间和累积到结果中的处理时间的部分。对于示例2-1中的管道，在经典批处理引擎上执行时，其外观类似于图2-3中所示。

图2-3 经典批处理

因为这是批处理管道，所以它会累积状态，直到看到所有输入（由顶部的绿色虚线表示）为止，此时将产生其单个输出48。在此示例中，我们正在计算总和 在整个事件时间内，因为我们没有应用任何特定的窗口转换；因此，用于状态和输出的矩形覆盖了整个x轴。但是，如果我们要处理无限制的数据源，那么经典的批处理是不够的；我们等不及输入结束，因为它实际上永远不会结束。我们想要的概念之一就是在第1章中介绍的窗口化。因此，在第二个问题（“事件时间在哪里计算结果？”）的上下文中，我们现在将简要地回顾窗口化。

哪里：开窗

如第1章所述，窗口化是沿时间边界分割数据源的过程。常见的开窗策略包括固定窗口，滑动窗口和会话窗口，如图2-4所示。

图2-4 示例窗口策略。为每个示例显示了三个不同的键，突出显示了对齐窗口（适用于所有数据）和未对齐窗口（适用于数据的子集）之间的差异。

为了更好地了解实际情况下的窗口化，让我们采用整数求和管道并将其窗口化为固定的两分钟窗口。使用Beam所做的更改是对Window.into转换的简单添加，您可以在示例2-2中看到该转换。

示例2-2 窗口求和代码

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES)))
 .apply(Sum.integersPerKey());

回想一下，Beam提供了一个在批处理和流处理中都可以使用的统一模型，因为语义上的批处理实际上只是流的一个子集。因此，我们首先在批处理引擎上执行此管道；机制更加简单明了，这将为我们提供一些可以直接与流媒体引擎进行比较的东西。图2-5给出了结果。

图2-5 批处理引擎上的窗口求和

和以前一样，输入会累积状态直到完全消耗掉，然后才产生输出。但是，在这种情况下，对于四个相关的两分钟事件时间窗口中的每个，我们得到四个输出：一个输出，而不是一个输出。
至此，我们已经回顾了我在第1章中介绍的两个主要概念：事件时域与处理时域之间的关系以及窗口化。如果想进一步，我们需要开始添加本节开头提到的新概念：触发器，水印和累积。

流播放：何时与如何

我们只是观察到批处理引擎上窗口化管道的执行。但是，理想情况下，我们希望降低结果延迟，并且还希望原生处理无限制的数据源。切换到流引擎是朝着正确方向迈出的一步，但是我们以前的等待直到我们的输入全部消耗完才能生成输出的策略不再可行。输入触发器和水印。

何时：关于触发器的奇妙之处是触发器是奇妙的事情！

触发器提供了以下问题的答案：“何时在处理时间内实现结果？”触发器声明窗口的输出何时应在处理时间内发生（尽管触发器本身可能会根据其他时域中发生的事情（例如事件时域中的水印进展）做出决定，片刻）。窗口的每个特定输出都称为窗口的窗格。

尽管可以想象到很多种可能的触发语义，但从概念上讲，只有两种通常有用的触发类型，而实际应用几乎总是使用一种或两种组合来解决：

重复更新触发器

这些随着窗口内容的发展而定期生成窗口的更新窗格。这些更新可以在每条新记录中实现，也可以在某些处理时间延迟（例如每分钟一次）之后进行。重复更新触发的周期选择主要是在平衡等待时间和成本方面的一种练习。

完整性触发

仅在确信该窗口的输入已完成到某个阈值之后，这些实体化窗口的窗格。这种触发器最类似于我们在批处理中所熟悉的触发器：只有在输入完成后，我们才提供结果。基于触发器的方法的不同之处在于，完整性的概念仅限于单个窗口的上下文，而不是始终限于整个输入的完整性。

重复更新触发器是流系统中遇到的最常见的触发器类型。它们易于实现且易于理解，并且为特定类型的用例提供了有用的语义：对物化数据集的重复（最终一致）更新，类似于您在数据库世界中对物化视图所获得的语义。

完整性触发器较少见，但提供的流语义与经典批处理世界中的流语义更加一致。它们还提供了用于推理数据丢失和后期数据之类的工具，我们将在探索驱动完整性触发的基础原语（水印）时简短地讨论这些问题（在下一章中）。

但首先，让我们开始简单一些，看看实际的一些基本重复更新触发器。为了使触发器的概念更加具体，让我们继续，然后将最直接的触发器类型添加到示例管道中：每个新记录都会触发的触发器，如示例2-3所示。

示例2-3 每条记录重复触发

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(Repeatedly(AfterCount(1))));
 .apply(Sum.integersPerKey());

如果我们要在流引擎上运行这个新管道，结果将类似于图2-6所示。

图2-6 流引擎上的每条记录触发

您可以看到我们现在如何为每个窗口获得多个输出（窗格）：每个对应的输入一次。当将输出流写入某种类型的表中时，这种触发模式效果很好，您可以简单地查询结果。每次查看表格时，您都会看到给定窗口的最新值，并且随着时间的推移，这些值将趋于正确。

每条记录触发的一个缺点是它非常健谈。在处理大规模数据时，汇总之类的聚合提供了一个很好的机会来减少流的基数而又不会丢失信息。这在您拥有大容量按键的情况下尤其明显。在我们的示例中，庞大的团队拥有许多活跃的参与者。想象一下一个大型多人游戏，其中将玩家分为两个派系之一，并且您希望按阵营统计统计数据。给定派系中每个玩家的每个新输入记录，都不必更新您的记数法。相反，您可能很乐意在经过一些处理时间延迟之后（例如每秒或每分钟）更新它们。使用处理时间延迟的一个不错的副作用是，它在大容量键或窗口上具有均衡作用：最终的流最终在基数方面更加统一。

触发器中有两种不同的处理时间延迟方法：对齐的延迟（其中延迟将处理时间切成在键和窗口之间对齐的固定区域）和未对齐的延迟（其中延迟与给定窗口中观察到的数据有关） ）。延迟未对齐的管道可能类似于示例2-4，其结果如图2-7所示。

示例2-4 在两分钟的处理时间边界上触发

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(Repeatedly(AlignedDelay(TWO_MINUTES)))
 .apply(Sum.integersPerKey());

图2-7 两分钟对齐的延迟触发器（即微批处理）

这种对齐的延迟触发实际上是您从微批量流传输系统（例如Spark Streaming）中获得的。关于它的好处是可预测性。您会同时在所有修改过的窗口上获得定期更新。这也是不利的一面：所有更新会同时发生，这会导致工作负载突发，通常需要更大的高峰调配才能正确处理负载。替代方法是使用未对齐的延迟。这看起来像Beam中的示例2-5。图2-8给出了结果。

示例2-5 在未对齐的两分钟处理时间边界上触发

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(Repeatedly(UnalignedDelay(TWO_MINUTES))
 .apply(Sum.integersPerKey());

图2-8 两分钟未对齐的延迟触发器

将图2-8中未对齐的延迟与图2-6中未对齐的延迟进行比较，很容易看出未对齐的延迟如何在整个时间范围内更均匀地分散负载。在任何给定窗口中涉及的实际延迟在两者之间有所不同，有时更多，有时更少，但最终平均延迟将基本保持不变。从这个角度来看，未对齐的延迟通常是大规模处理的更好选择，因为它们会导致随时间推移的负载分配更加均匀。

重复更新触发器非常适合用例，在这些用例中，我们只是希望随着时间的推移定期对结果进行定期更新，并且这些更新趋于正确，而没有明确指示何时达到正确性的情况就很好。但是，正如我们在第1章中讨论的那样，在事件发生的时间与管道实际观察到的时间之间，分布式系统的变化不定通常会导致不同程度的偏斜，这意味着很难推断何时输出呈现您输入数据的准确和完整的视图。对于输入完整性很重要的情况，重要的是要有某种方式来推理完整性，而不是一味地信任由偶然发现到您的管道中的任何数据子集计算出的结果。输入水印。

何时：水印

水印是该问题答案的一个支持方面：“何时在处理中实现结果？” 水印是事件时域中输入完整性的时间概念。换句话说，它们是系统相对于事件流中处理的记录的事件时间（有界或无界，尽管其用途在无界情况下更为明显）衡量进度和完整性的方式。

回顾一下第1章中的图，在图2-9中进行了稍微修改，其中我将事件时间和处理时间之间的时差描述为大多数现实世界中分布式数据处理系统的时间不断变化的函数。

图2-9 事件时间进度，偏斜和水印

我声称代表真实的那条蜿蜒的红线本质上是水印。随着处理时间的推移，它捕获了事件时间完整性的进度。从概念上讲，您可以将水印视为函数F（P）→E，它在处理时间上花费了一点，在事件时间上返回了一点。事件时间E的那一点是系统认为观察到事件时间小于E的所有输入的时间点。换句话说，这是断言不会再看到事件时间小于E的更多数据。根据水印的类型（完美或启发式），断言可以分别是严格的保证或有根据的猜测：

完美的水印

如果我们对所有输入数据都有完备的知识，则可以构造出完美的水印。在这种情况下，就没有迟到的数据。所有数据都是提早或准时的。

启发式水印

对于许多分布式输入源，完全了解输入数据是不切实际的，在这种情况下，下一个最佳选择是提供启发式水印。启发式水印使用有关输入的任何可用信息（分区，分区内的排序（如果有），文件的增长率等）来提供尽可能准确的进度估计。在许多情况下，此类水印的预测可以非常准确。即使这样，使用启发式水印也意味着它有时可能是错误的，这将导致数据过时。我们向您介绍如何尽快处理最新数据。

因为水印相对于我们的输入提供了完整性的概念，所以水印构成了前面提到的第二种触发器的基础：完整性触发器。水印本身是一个引人入胜且复杂的主题，您将在第3章中深入了解Slava的水印。但是，现在，让我们通过更新示例管道以利用基于水印的完整性触发器来实际研究它们，如实施例2-6所示。

示例2-6 水印完整性触发

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(AfterWatermark()))
 .apply(Sum.integersPerKey());

现在，水印的一个有趣的特性是它们是一类函数，这意味着有多个不同的函数F（P）→E可以不同程度地满足水印的特性。如前所述，对于您非常了解输入数据的情况，可能会构建一个完美的水印，这是理想的情况。但是，如果您缺乏对输入的完全了解，或者由于计算量太高而无法计算出完美的水印，则可以选择使用启发式来定义水印。我想在此说明的是，给定的水印算法在使用中与管道本身无关。我们将不在此处详细讨论实施水印的含义（Slava在第3章中进行了说明）。现在，为了帮助理解给定输入集可以应用不同水印的想法，让我们看一下示例2-6中的管道，该管道在同一数据集上执行但使用两个不同的水印实现时（图2- 10）：在左侧，是完美的水印；右边是启发式水印。

在这两种情况下，当水印通过窗口的末端时，窗口都会变为实物。如您所料，完美的水印可以随着时间的推移完美地捕获管道的事件时间完整性。相比之下，右侧启发式水印所使用的特定算法未考虑值9，5会从输出延迟和正确性方面彻底改变物化输出的形状（如错误答案所示）为[12:00，12:02）窗口提供的5分之一）。

图2-9中的水印触发器与我们在图2-5至2-7中看到的重复更新触发器之间的最大区别在于，水印为我们提供了一种推断输入完整性的方法。在系统实现给定窗口的输出之前，我们知道系统尚未相信输入是完整的。这对于要推理输入中缺少数据或缺少数据的用例尤其重要。

图2-10 具有完美（左）和启发式（右）水印的流引擎上的窗口求和

外部联接是丢失数据用例的一个很好的例子。没有水印之类的完整性概念，您如何知道何时放弃并发出部分联接，而不是继续等待该联接完成？你没有而且基于处理时间延迟的决定（这是缺乏真正的水印支持的流系统中的常见方法）不是安全的方法，因为我们在第1章中谈到的事件时间偏斜具有可变性。只要偏斜保持小于选定的处理时间延迟，您的丢失数据结果将是正确的，但是只要偏斜超过该延迟，它们就会突然变得不正确。从这个角度来看，事件时间水印是许多现实世界中流媒体用例的一个关键难题，这些用例必须要推理输入中缺少数据，例如外部联接，异常检测等。
话虽如此，这些水印示例还突出显示了水印的两个缺点（以及完整性的任何其他概念），特别是它们可以是以下之一：

太慢了

当任何类型的水印由于已知的未处理数据而被正确延迟（例如，由于网络带宽限制而导致输入日志缓慢增长）时，如果水印的前进是刺激结果的唯一依据，则直接转化为输出延迟。

这在图2-10的左图中最明显，对于最晚到达的9，即使所有窗口的输入数据都较早完成，迟到的9也会阻止所有后续窗口的水印。这对于第二个窗口[12:02，12:04）尤其明显，从窗口中的第一个值出现直到我们看到该窗口的任何结果为止，要花费将近七分钟的时间。在此示例中，启发式水印不会遇到同样严重的问题（直到输出前五分钟），但这并不是说启发式水印永远不会遭受水印滞后的困扰；这实际上只是我在此特定示例中选择从启发式水印中省略的记录的结果。

这里的重点如下：尽管水印提供了非常有用的完整性概念，但是从延迟角度来看，依靠完整性来产生输出通常并不理想。想象一下一个仪表板，其中包含按小时或天显示的重要指标。您不太可能要等待整整一个小时或一天才能开始看到当前窗口的结果；这是使用经典批处理系统为此类系统提供动力的痛苦点之一。取而代之的是，随着输入的发展并最终变得完整，看到这些窗口的结果随着时间的推移而细化会更好。

太快

如果启发式水印的发布错误早于应有的时间，则事件发生时间早于水印的数据可能会在稍后的某个时间到达，从而产生较晚的数据。这就是在右边的示例中发生的情况：在观察到该窗口的所有输入数据之前，水印超过了第一个窗口的末尾，导致输出值为5而不是14，这是不正确的。启发式水印问题；它们的启发性质意味着它们有时会出错。结果，如果您关心正确性，仅依靠它们来确定何时实现输出是不够的。

在第一章中，我做了一些相当强调的陈述，即完整性的概念对于大多数需要对无边界数据流进行有序无序处理的用例而言是不够的。这两个缺点（水印太慢或太快）是这些论点的基础。您根本无法从仅依赖完整性概念的系统中获得低延迟和正确性。那么，对于您确实想要两全其美的情况，一个人应该做什么？好吧，如果重复的更新触发器提供了低延迟的更新，但却无法推理出完整性，而水印提供了完整性的概念，但是可变且可能存在高延迟，那么为什么不将其功能结合在一起呢？

何时：提早/准时/晚触发FTW！

现在，我们研究了两种主要类型的触发器：重复更新触发器和完整性/水印触发器。在许多情况下，仅靠它们都不是足够的，但是将它们组合在一起就足够了。Beam通过提供标准水印触发器的扩展来识别这一事实，该扩展还支持水印任一侧的重复更新触发器。这被称为早期/按时间/延迟触发器，因为它将由复合触发器实现的窗格划分为三类：

• 零个或多个早期窗格，这是重复更新触发器的结果，该触发器定期触发直到水印通过窗口的结尾。这些触发产生的窗格包含推测性结果，但允许我们观察随着新输入数据的到达，窗口随时间的变化。这弥补了水印有时太慢的缺点。
• 一个单一的按时间窗格，这是在水印通过窗口结尾之后触发完整性/水印触发器的结果。触发是特殊的，因为它提供了一个断言，即系统现在认为此窗口的输入已完成。这意味着现在可以安全地推断出数据丢失了；例如，在执行外部联接时发出部分联接。
• 零个或多个后期窗格，这是另一个（可能是不同的）重复更新触发器的结果，该触发器会在水印经过窗口结尾之后的任何时候，在迟到数据到达时定期触发。在完美水印的情况下，后期窗格将始终为零。但是在启发式水印的情况下，水印未能正确说明的任何数据都会导致延迟触发。这弥补了水印太快的缺点。

让我们看看它的实际效果。我们将更新管道，以使用定期处理时间触发器，并针对早期触发使用一分钟的对齐延迟，并针对后期触发使用按记录的触发。这样，早期触发将为我们提供大量的大批量窗口批处理（由于触发器每分钟仅触发一次，而不管窗口的吞吐量如何），但是我们不会引入不必要的延迟 对于后期触发，如果我们使用合理准确的启发式水印，希望这种情况很少见。在Beam中，其外观类似于示例2-7（图2-11显示了结果）。

示例2-7 通过早期/及时/后期API提前，按时和延迟触发

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(AfterWatermark()
 .withEarlyFirings(AlignedDelay(ONE_MINUTE))
 .withLateFirings(AfterCount(1))))
 .apply(Sum.integersPerKey());

图2-11 流引擎上的窗口求和，具有提前，按时和延迟触发

这个版本比图2-9有两个明显的改进：

• 对于第二个窗口中的“水印太慢”情况，[12:02，12:04）：现在，我们每分钟提供一次定期的早期更新。在完美的水印情况下，差异最为明显，因为这种情况下，首次输出的时间从将近7分钟减少到了3分半左右。但是在启发式情况下也明显得到了改进。现在，这两个版本都可以随时间提供稳定的细化（值分别为7、10，然后为18的窗格），在输入变得完整和窗口的最终输出窗格实现之间的等待时间相对较短。
• 对于第一个窗口中的“启发式水印太快”的情况，[12:00，12:02）：当值9出现较晚时，我们立即将其合并到一个新的校正后的值为14的窗格中。

这些新触发器的一个有趣的副作用是，它们可以有效地规范完美和启发式水印版本之间的输出模式。尽管图2-10中的两个版本完全不同，但此处的两个版本看起来非常相似。它们看起来也更类似于图2-6至2-8中的各种重复更新版本，但有一个重要的区别：由于使用了水印触发器，我们还可以推断出在生成的结果中输入完整性。提早/准时/延迟触发。这使我们可以更好地处理关心丢失数据的用例，例如外部联接，异常检测等。

完美和启发式的早期/按时间/后期版本之间最大的剩余差异是窗口生存期界限。在理想的水印情况下，我们知道在水印通过窗口结束之后再也看不到窗口的任何数据，因此我们可以在那时删除窗口的所有状态。在启发式水印的情况下，我们仍然需要在一段时间内保持窗口状态以解释较晚的数据。但是到目前为止，我们的系统尚无任何好的方法来知道每个窗口需要保持多长时间。那就是允许迟到的地方。

何时：允许延迟（即垃圾回收）

在继续讨论最后一个问题（“结果的优化有何关系？”）之前，我想谈谈长期存在的无序流处理系统中的实际必要性：垃圾收集。在图2-11的启发式水印示例中，每个窗口的持久状态在示例的整个生命周期中都徘徊；这是必要的，以使我们能够/在它们到达时适当地处理较晚的数据。但是，尽管能够一直保持所有持久状态直到时间结束是很棒的，但实际上，当处理无限制的数据源时，对于给定窗口无限期地保持状态（包括元数据）通常是不现实的 ;我们最终会用完磁盘空间（或者至少要花钱，因为随着时间的推移，旧数据的价值会逐渐减少）。

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

由于允许的延迟和水印之间的相互作用有些微妙，因此值得一看。让我们以示例2-7 /图2-11中的启发式水印管道为例，并在示例2-8中添加一分钟的延迟范围（请注意，严格选择此特定范围是因为它很好地适合了图表；对于实际- 在世界范围内的用例中，更大的视野可能会更加实用）：

示例2-8 提前/准时/延迟点火并允许延迟

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(
 AfterWatermark()
 .withEarlyFirings(AlignedDelay(ONE_MINUTE))
 .withLateFirings(AfterCount(1)))
 .withAllowedLateness(ONE_MINUTE))
 .apply(Sum.integersPerKey());

该管道的执行类似于图2-12，其中我添加了以下功能来突出显示允许延迟的影响：

• 现在，用勾号注释表示处理时间当前位置的黑色粗线，该勾号指示所有活动窗口的延迟时间（事件时间）。
• 当水印通过窗口的最新地平线时，该窗口将关闭，这意味着该窗口的所有状态都将被丢弃。我留下了一个虚线矩形，显示了关闭窗口时（在两个域中）的时间范围，一条小尾巴向右延伸，表示该窗口的延迟水平（与水印进行对比）。
• 仅对于该图，我为第一个窗口添加了一个额外的延迟基准，其值为6。6延迟了，但仍在允许的延迟范围内，因此将其合并到值为11的更新结果中。然而，9 到达延迟范围之外，因此将其丢弃。

图2-12 允许延迟及早/准时/晚点火

关于延迟时间的最后两点注释：

• 绝对清楚，如果您碰巧正在使用可提供完美水印的来源的数据，则无需处理延迟数据，并且允许的延迟时间范围为零秒是最佳的。这就是我们在图2-10的完美水印部分中看到的。
• 即使在使用启发式水印时，也需要指定延迟时间范围的规则的一个值得注意的例外是，例如始终计算有限数量的密钥的全局聚合（例如，计算对您的访问的总次数） 网站上的所有网站（按网络浏览器系列分组）。在这种情况下，系统中活动窗口的数量受到使用中有限键空间的限制。只要密钥数量保持在可控制的低水平，就无需担心通过允许的延迟来限制窗口的寿命。

实用性已经足够，让我们继续第四个也是最后一个问题。

如何：积累

当使用触发器在一段时间内为单个窗口生成多个窗格时，我们发现自己面临最后一个问题：“结果的细化如何关联？”在到目前为止我们看到的示例中，每个连续的窗格都基于紧接其前的窗格。但是，实际上有三种不同的累积模式：

丢弃

每次实现窗格时，任何存储状态都将被丢弃。这意味着每个连续的窗格都独立于之前的窗格。当下游使用者自己进行某种积累时，丢弃模式很有用。例如，当将整数发送到希望接收增量的系统中时，它将相加在一起以产生最终计数。

积累中

如图2-6至2-11所示，每次实现窗格时，任何存储的状态都会保留，将来的输入会累积到现有状态中。这意味着每个连续的窗格都建立在先前的窗格上。当以后的结果可以简单地覆盖以前的结果时，例如在将输出存储在HBase或Bigtable之类的键/值存储中时，累积模式很有用。

累积与缩回

这类似于累积模式，但是在生成新窗格时，它还会为以前的窗格生成独立的收缩。回缩（与新的累加结果结合使用）本质上是一种明确的说法，“我之前告诉过您结果是X，但我错了。摆脱我上次告诉您的X，然后将其替换为Y。”在两种情况下，撤回特别有用：

• 当下游的使用者按不同的维度对数据进行重新分组时，新值很有可能最终以与先前值不同的键键入，因此最终以不同的组结束。在这种情况下，新值不能只覆盖旧值；相反，您需要撤回以删除旧值
• 使用动态窗口（例如，我们将在稍后讨论的会话）时，由于窗口合并，新值可能会替换多个以前的窗口。在这种情况下，可能难以仅从新窗口确定要替换哪些旧窗口。对旧窗口进行显式缩回可以使任务简单明了。我们将在第8章中详细看到一个示例。

并排观察时，每个组的不同语义更加清晰。考虑图2-11中第二个窗口的两个窗格（一个事件时间范围为[12:06，12:08]的窗格）（一个有早/准时/延迟触发器的窗格）。表2-1显示了在三种累加模式下每个窗格的值是什么样的（累加模式是图2-11本身使用的特定模式）。

表2-1 使用图2-11中的第二个窗口比较累积模式

让我们仔细看看发生了什么：

丢弃

每个窗格仅包含在该特定窗格中到达的值。因此，观察到的最终值不能完全捕获总和。但是，如果要对所有独立窗格本身求和，将得出正确的答案12。这就是为什么当下游使用者自己在物化窗格上执行某种聚合时，丢弃模式很有用的原因。

积累中

如图2-11所示，每个窗格都包含在该特定窗格中到达的值，以及先前窗格中的所有值。这样，观察到的最终值正确地捕获了总计12的总和。但是，如果您要对各个窗格本身进行汇总，那么实际上您将对来自窗格1的输入进行重复计数，从而得出错误的总和为15这就是为什么当您可以简单地用新值覆盖以前的值时，累积模式最有用的原因：新值已经合并了到目前为止看到的所有数据。

累积与缩回

每个窗格都包含新的累积模式值以及上一个窗格值的缩回。这样，观察到的最后一个值（不包括收缩）以及所有物化窗格的总和（包括收缩）都为您提供了正确的答案12。这就是为什么收缩如此强大的原因。

例2-9演示了实际的丢弃模式，说明了我们将对例2-7进行的更改：

示例2-9 早期/准时/延迟点火的丢弃模式版本

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(
 AfterWatermark()
 .withEarlyFirings(AlignedDelay(ONE_MINUTE))
 .withLateFirings(AtCount(1)))
 .discardingFiredPanes())
 .apply(Sum.integersPerKey());

在具有启发式水印的流引擎上再次运行将产生如图2-13所示的输出。

图2-13 流引擎上的早期/准时/延迟点火的丢弃模式版本

即使输出的总体形状类似于图2-11中的累积模式版本，也请注意此丢弃版本中的窗格如何不重叠。结果，每个输出彼此独立。

如果我们想看看撤退的实际效果，那么变化将是相似的，如示例2-10所示。结果如图2-14所示。

示例2-10 早/准/晚点火的累积和缩回模式版本

PCollection<KV<Team, Integer>> totals = input
 .apply(Window.into(FixedWindows.of(TWO_MINUTES))
 .triggering(
 AfterWatermark()
 .withEarlyFirings(AlignedDelay(ONE_MINUTE))
 .withLateFirings(AtCount(1)))
 .accumulatingAndRetractingFiredPanes())
 .apply(Sum.integersPerKey());

图2-14 流引擎上早/晚触发的累积和收缩模式版本

由于每个窗口的窗格都重叠，因此要清楚地看到缩回有点棘手。缩进以红色表示，它与重叠的蓝色窗格结合在一起产生略带紫色的颜色。我还在给定窗格中水平移动了两个输出的值（并用逗号隔开），以使其易于区分。

图2-15并排组合了图2-9，图2-11（仅启发式）和图2-14的最终帧，从而提供了三种模式的良好视觉对比。

图2-15 累积模式的并排比较

您可以想象，按顺序显示的模式（丢弃，累加，累加和收回）在存储和计算成本方面依次昂贵。为此，累积模式的选择提供了另一个维度，可以沿着正确性，等待时间和成本的轴进行权衡。

概要

通过本章的学习，您现在已经了解了鲁棒的流处理的基础知识，并准备进入世界并做一些令人惊奇的事情。当然，还有八章急切地等待着您的关注，因此希望您不会像现在这样紧迫地走下去。但是无论如何，让我们回顾一下我们刚刚介绍的内容，以免您在匆忙前进时忘记了其中的任何内容。首先，我们涉及的主要概念：

事件时间与处理时间

事件何时发生以及数据处理系统何时观察到事件之间的最重要区别。

加窗

通过沿时间边界切片无边界数据来管理无边界数据的常用方法（在处理时间或事件时间内，尽管我们将Beam Model中的窗口定义缩小为仅在事件时间内表示）。

触发器

一种声明式机制，用于精确指定何时实现输出对您的特定用例有意义。

水印

事件时间进度的强大概念，为在无界数据上运行的无序处理系统中提供了一种推理完整性（并因此丢失数据）的方法。

积累

对于单个窗口而言，随着结果的演变多次实现，结果的细化之间的关系。

其次，我们用来构架探索的四个问题：

• 计算出什么结果？ =转换。
• 结果在哪里计算？ =开窗。
• 处理时间何时实现？ =触发器加水印。
• 结果的细化如何关联？ =积累。

第三，为获得这种流处理模型所提供的灵活性（最后，实际上就是这一切：平衡正确性，延迟和成本等竞争压力），回顾一下我们在输出方面的主要变化只需最少的代码更改即可在同一数据集上实现的功能：

综上所述，到目前为止，我们仅考察了一种类型的窗口：事件时间中的固定窗口。众所周知，窗口化有很多维度，在我们将其称为“光束模型”之前，我想再谈谈至少两个维度。但是，首先，我们将走一点弯路，以更深入地研究水印世界，因为这种知识将有助于构成未来的讨论（并使其本身引人入胜）。进入Slava，右上…