分布式 blog

记录技术点滴

0%

《Streaming System》-第二章： 数据处理的什么、何地、何时以及如何进行[完整]

发表于 更新于 分类于

目录#

好了，派对人们，是时候开始具体化了！

第一章主要涉及三个方面：术语，准确定义当我使用过载术语“流式处理”时的含义；批处理与流处理，比较两种系统的理论能力，并推断只有两个方面是将流处理系统推向超越批处理系统所必须的：正确性和关于时间的推理工具；以及数据处理模式，研究在处理有界数据和无界数据时，批处理系统和流处理系统采用的概念方法。

在本章中，我们将进一步关注第一章中的数据处理模式，但更详细地在具体示例的背景下进行。到最后，我们将涵盖我认为对于强大的乱序数据处理所必需的核心原则和概念；这些是真正将您推向经典批处理之外的关于时间的推理工具。

为了让您了解实际情况，我使用Apache Beam代码片段，配合时间流逝图表，提供概念的可视化表示。Apache Beam是一种统一的批处理和流处理编程模型和可移植性层，具有各种语言的具体SDK（例如Java和Python）。使用Apache Beam编写的管道可以在任何支持的执行引擎上进行便携式运行（Apache Apex、Apache Flink、Apache Spark、Cloud Dataflow等）。

我在这里使用Apache Beam作为示例，不是因为这是一本Beam书籍（它不是），而是因为它最完全地体现了本书中描述的概念。回想一下，“流处理102”最初编写时（当它仍然是Google Cloud Dataflow的Dataflow模型，而不是Apache Beam的Beam模型时），它是唯一存在的系统，提供了所有我们将在此处涵盖的示例所需的表达能力。一年半后，我很高兴地说，很多事情发生了改变，大多数主要系统已经移动或正在朝向支持类似于本书所描述的模型。因此，请放心，尽管是通过Beam的镜头，我们在这里涵盖的概念同样适用于您将遇到的大多数其他系统。

路线图#

为了帮助为本章打下基础，我想提出五个主要概念，这些概念将支撑所有讨论，以及大部分第一部分的内容。我们已经讨论了其中两个。

在第一章中，我首先建立了事件时间（事件发生的时间）和处理时间（在处理期间观察到的时间）之间的关键区别。这为本书提出的一个主要论点奠定了基础：如果您关心正确性和事件实际发生的上下文，则必须相对于它们固有的事件时间而不是在分析过程中遇到它们的处理时间来分析数据。

然后，我介绍了窗口化的概念（即沿着时间边界对数据集进行分区），这是一种常用的方法，用于应对无限数据源在技术上可能永远不会结束的事实。窗口化策略的一些更简单的例子是固定和滑动窗口，但更复杂的窗口化类型，例如会话（其中窗口是由数据本身的特征定义的;例如，捕获用户的每个活动会话，然后是不活动的间隙），也广泛使用。

除了这两个概念之外，我们现在将仔细研究另外三个概念：

  • 触发器 Triggers

    触发器是一种声明窗口的输出相对于某些外部信号何时被实现的机制。 触发器提供了选择何时发出输出的灵活性。在某种意义上，您可以将它们视为用于指示何时应该实现结果的流量控制机制。 另一种看法是，触发器就像相机的快门释放，允许您声明何时以时间为快照记录正在计算的结果。
    触发器还可以观察窗口多次演变时的输出，从而打开了随着时间推移改进结果的大门，这允许在数据到达时提供推测结果，以及处理上游数据（修订）随时间变化或数据到达延迟的情况（例如，移动场景，其中某人的手机在离线时记录各种操作和事件时间，然后在恢复连接时上传这些事件进行处理）。

  • 水位线 Watermarks

    水位线是相对于事件时间的输入完整性的概念。具有时间 X 值的水位线表明：“事件时间小于X的所有输入数据都已被观察到。”因此，水位线在观察没有已知结束的无限数据源时作为进度的指标。我们在本章中介绍了水位线的基础知识，然后Slava在第3章中深入探讨了这个主题。

  • 累积 Accumulation

    累积模式指定了对于同一窗口所观察到的多个结果之间的关系。这些结果可能完全不相关，即代表随时间独立的增量，也可能存在重叠。不同的累积模式具有不同的语义和相关成本，因此适用于各种用例。

此外，因为我认为这使得更容易理解所有这些概念之间的关系，我们重新审视了旧的并在回答四个问题的结构内探索了新的，我提出这些问题对于每个无限数据处理问题都至关重要：

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

我们在本书的其余部分中更详细地研究了这些问题。是的，我将尝试将这个颜色方案运用到底，以尝试清晰地表明哪些概念与What/Where/When/How习语中的哪些问题有关系。你是受欢迎的 <winky smiley/>。

批处理基础知识：什么 和 何地#

好的，让我们开始吧。首先，是批处理。

什么：转换#

应用在经典批处理中的转换回答了这个问题：“计算出了什么结果？”尽管你可能已经熟悉经典批处理，但我们还是要从这里开始，因为它是我们添加所有其他概念的基础。

在本章的其余部分（事实上，在本书的很大一部分中），我们将看到一个单一的示例：在由九个值组成的简单数据集上计算有键[keyed]整数总和。假设我们编写了一个团队游戏，并且我们想要构建一个管道，通过总结用户手机报告的个人分数来计算团队分数。如果我们将我们的九个示例得分捕获在名为“UserScores”的SQL表中，它可能看起来像这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
----------------------------------
|名称|团队|分数|事件时间|ProcTime|
----------------------------------
|朱莉|TeamX|5|12:00:26|12:05:19|
|弗兰克|TeamX|9|12:01:26|12:08:19|
|爱德华|TeamX|7|12:02:26|12:05:39|
|朱莉|TeamX|8|12:03:06|12:07:06|
|艾米|TeamX|3|12:03:39|12:06:13|
|弗雷德|TeamX|4|12:04:19|12:06:39|
|纳奥米|TeamX|3|12:06:39|12:07:19|
|贝基|TeamX|8|12:07:26|12:08:39|
|纳奥米|TeamX|1|12:07:46|12:09:00|
----------------------------------

请注意，此示例中的所有分数均来自同一团队的用户；这是为了让示例保持简单，因为我们的图表中的维度数量有限。而且，因为我们按团队分组，所以我们只关心最后三列：

  • 分数
    与此事件相关联的单个用户分数
  • 事件时间[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有两个基本原语：

  • PCollections

    这些表示可以执行并行转换的数据集（因此名称的开头有“P”）。

  • PTransforms

    这些应用于PCollections以创建新的PCollections。

    PTransforms可以执行逐个元素的转换，它们可以将多个元素分组/聚合在一起，或者它们可以是其他PTransforms的组合，如图2-2所示。

图2-2.转换类型

对于我们的示例，我们通常假设我们从预加载的PCollection<KV<Team，Integer>>“input”（即由Team和Integers组成的键/值对的PCollection，其中Teams仅是表示团队名称的字符串，而Integers是来自相应团队的任何个人的分数）开始。在实际的管道中，我们将通过从I / O源读取原始数据（例如，日志记录）的PCollection并将其转换为PCollection<KV<Team，Integer>>来获取输入。通过将日志记录解析为适当的键/值对。为了在这个第一个示例中清楚起见，我包括了所有这些步骤的伪代码，但在接下来的示例中，我省略了I / O和解析。

因此，对于一个简单地从I / O源读取数据，解析团队/分数对并计算每个团队的总分的管道，我们将有像示例2-1中所示的那样的东西。

示例2-1.总和管道

1
2
3
4
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章中介绍的。因此，在我们的第二个问题“事件时间的哪里计算结果？”的上下文中，我们现在将简要重新访问窗口。 因此，在我们的第二个问题“事件时间在哪里计算结果？”的背景下，我们现在将简要重温窗口化。

何地(Where): 窗口 #

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

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

为了更好地了解实践中的窗口处理，让我们将整数求和管道分隔成固定的两分钟窗口。
使用Beam，改变很简单，只需添加一个Window.into转换，您可以在示例2-2中看到高亮显示。

示例2-2.窗口化的求和代码

1
2
3
PCollection<KV<Team，Integer>> totals = input
  .apply（Window.into（FixedWindows.of（TWO_MINUTES）））
  .apply（Sum.integersPerKey（））;

请记住，Beam提供了一个统一的模型，可以在批处理和流处理中使用，因为从语义上讲，批处理实际上只是流处理的一个子集。因此，让我们首先在批处理引擎上执行此管道；机制更加简单，当我们切换到流引擎时，它将为我们提供直接比较的内容。图2-5呈现了结果。

图2-5.在批处理引擎上的分段求和

与以前一样，输入会在状态中累积，直到完全消耗，此后才会产生输出。但是，在这种情况下，我们不仅仅得到一个输出，而是得到四个输出：每个相关的两分钟事件时间窗口的单个输出。

到目前为止，我们已经重新访问了我在第1章中介绍的两个主要概念：事件时间和处理时间域之间的关系以及窗口。如果我们想继续，我们将需要开始添加本节开头提到的新概念：触发器、水位线和累积。

流处理：何时(When) 和 如何转换(How) #

我们刚刚观察了在批处理引擎上执行窗口化管道的过程。但是，理想情况下，我们希望结果具有更低的延迟，并且我们还希望本地处理无界的数据源。切换到流处理引擎是朝着正确方向迈出的一步，但是我们之前等待输入完全被消耗后才生成输出的策略已经不再可行。于是，我们引入了触发器和水印的概念。

何时：关于触发器的美妙之处#

就是触发器是美妙的东西！#

触发器提供了答案：在处理时间中“何时” 来实现结果材料化。触发器声明何时在处理时间中应该发生窗口的输出（尽管触发器本身可能基于发生在其他时间域（例如事件时间域中进行的水位线进展）的事情做出这些决定，正如我们将在接下来的几个时刻中看到的）。每个窗口的特定输出称为窗格。

虽然可以想象出相当广泛的可能触发语义，但在概念上只有两种通常有用的触发器类型，实际应用几乎总是使用其中一种或两种的组合：

  • 重复更新触发器
    这些触发器会定期生成窗口的更新窗格，随着其内容的变化而变化。这些更新可以在每个新记录到来时进行实现，也可以在一定的处理时间延迟之后进行，例如每分钟一次。选择重复更新触发器的周期主要是在平衡延迟和成本之间进行权衡。

  • 完整性触发器
    这些触发器仅在相信窗口的输入已经完整达到某个阈值时才为窗口创建一个窗格。这种类型的触发器最类似于我们在批处理中熟悉的处理方式：只有在输入完成之后才会提供结果。触发器方法中的区别在于完整性的概念仅限于单个窗口的上下文范围，而不总是绑定到整个输入的完整性。

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

完整性触发器很少遇到，但提供与经典批处理世界中更相似的流式语义。它们还提供了用于推理诸如缺失数据和延迟数据之类的事物的工具，我们很快就会讨论（并在下一章中）当我们探索驱动完整性触发器的基础原语时：水位线。

但首先，让我们从简单的方面入手，看看一些基本的重复更新触发器的操作。为了使触发器的概念更加具体化，让我们继续向我们的示例管道添加最简单的触发器类型：每次新记录都会触发的触发器，如示例2-3所示。

示例2-3。每个记录都会重复触发

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

如果我们在流式引擎上运行这个新管道，结果会像图2-6所示。

图2-6.在流处理引擎上的单个记录触发

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

单个记录触发的一个缺点是它非常繁琐。在处理大规模数据时，聚合（例如求和）提供了一个很好的机会，可以在不丢失信息的情况下减少流的基数。这对于具有高交易量密钥的情况特别明显；对于我们的示例，具有许多活动玩家的大型团队。想象一下一个大规模的多人游戏，其中玩家被分为两个阵营，并且您想按阵营基础上统计统计数据。对于给定阵营中的每个玩家的每个新输入记录更新您的统计数据可能是不必要的。相反，您可能会满意地在一些处理时间延迟之后进行更新，例如每秒或每分钟。使用处理时间延迟的好处是，它对高交易量键或窗口具有均等化效应：由此产生的流在基数方面最终会更加统一。

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

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

1
2
3
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。在不对齐的两分钟处理时间边界上触发

1
2
3
4
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章讨论的那样，分布式系统的不确定性通常会导致事件发生的时间与您的管道实际观察到该事件的时间之间的变化程度不同，这意味着很难理解您的输出何时提供了输入数据的准确和完整视图。对于重视输入完整性的情况，重要的是有某种方法来推理完整性，而不是盲目地信任哪个子集的数据已经流到您的管道中。然后进入水位线。

何时：水位线#

水位线是回答“处理时间何时实现结果”的问题的一个支持方面。水位线是输入完整性在事件时间域中的时间概念。换句话说，它们是系统相对于正在处理的事件流记录的事件时间测量进度和完整性的方式（有界或无界的情况下都很有用，但在无界的情况下它们更明显）。

回想一下第一章中的这个图表，在图2-9中稍作修改，我将事件时间和处理时间之间的偏差描述为大多数实际分布式数据处理系统中时间的不断变化的函数。

图2-9. 事件时间进度，偏差和水位线

我声称代表现实的那条蜿蜒的红线本质上是水位线；它捕获事件时间完整性随着处理时间的推进的进展。在概念上，您可以将水位线视为一个函数，F(P)→E，它将处理时间中的一个点返回到事件时间中的一个点。事件时间中的那个点，E，是系统认为已经观察到了所有事件时间小于E的输入的点。换句话说，它是一个断言，即不会再看到事件时间小于E的更多数据。根据水位线的类型，完美或启发式，该断言可以是严格保证或有教养的猜测。

  • 完美水位线

    对于我们拥有所有输入数据的情况，可以构建完美的水位线。在这种情况下，不存在延迟数据；所有数据都是提前或准时的。

  • 启发式水位线

    对于许多分布式输入源，实际上不可能完全了解输入数据，因此提供启发式水位线是下一个最佳选择。启发式水位线使用有关输入的任何信息（分区、如果有的话，分区内的排序、文件的增长率等），以提供尽可能准确的进度估计。在许多情况下，这样的水位线可以令人惊讶地准确预测。即便如此，使用启发式水位线意味着它有时可能是错误的，这将导致延迟数据。我们很快就会向您展示处理延迟数据的方法。

因为它们提供了相对于我们的输入的完整性概念，水位线是前面提到的第二种触发器的基础：完整性触发器。水位线本身是一个非常有趣和复杂的话题，当您到达第3章的Slava的水位线深度探索时，您就会看到。但是，现在让我们通过更新我们的示例管道以利用基于水位线的完整性触发器来看看它们的实际应用，如示例2-6所示。

示例2-6. 水位线完整性触发器

1
2
3
4
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的值，这大大改变了材料化输出的形状，无论是在输出延迟还是正确性方面（如为[12:00,12:02)窗口提供的不正确答案5所示）。水位线触发器与图2-9中所示的重复更新触发器之间的最大区别是，水位线使我们有一种方法来推断我们的输入的完整性。在系统为给定窗口实现输出之前，我们知道系统尚未认为输入已完成。这对于希望推断输入中缺少数据或缺少数据的用例尤其重要。

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

缺少数据用例的一个很好的例子是外连接。没有像水位线这样的完整性概念，您如何知道何时放弃发出部分连接而不是继续等待该连接完成？你不知道。在缺乏真正的水位线支持的流式系统中，将该决定基于处理时间延迟并不安全，因为我们在第1章中所说的事件时间偏差的可变性：只要偏差保持小于所选处理时间延迟，缺少数据的结果将是正确的，但是任何时候偏差超过该延迟，它们将突然变得不正确。从这个角度来看，事件时间水位线是许多必须推断输入中缺少数据（例如外连接，异常检测等）的实际流用例的关键部分。

现在，话虽如此，这些水位线示例也突出了水位线（和任何其他完整性概念）的两个缺点，具体而言，它们可能是以下两者之一：

  • 太慢

    当任何类型的水位线由于已知的未处理数据（例如由于网络带宽限制而慢慢增长的输入日志）而被正确延迟时，如果仅依赖于水位线推进来刺激结果，则会直接导致输出延迟。

    这在图2-10的左图中最为明显，由于晚到达的9为所有后续窗口的水位线保留，即使这些窗口的输入数据更早地变得完整。这对于第二个窗口[12:02,12:04)尤其明显，因为从窗口第一个值发生的时间到我们看到窗口的任何结果为止，需要将近七分钟的时间。在这个例子中使用的启发式水位线没有如此严重的问题（五分钟的输出），但是不要认为启发式水位线永远不会受到水位线滞后的影响；这实际上只是我选择从这个特定示例中省略启发式水位线的记录的结果。

    这里的重要观点如下：尽管水位线提供了非常有用的完整性概念，但是从延迟的角度来看，仅依赖于完整性来产生输出通常并不理想。想象一下包含按小时或按天分组的有价值指标的仪表板。您不太可能想要等待整个小时或天才开始看到当前窗口的结果；这是使用经典批处理系统来为这些系统提供动力的痛点之一。相反，随着输入的演变和最终变得完整，看到这些窗口的结果逐渐完善会更好。

  • 太快

    当启发式水位线被错误地提前时，事件时间早于水位线的数据可能会在一段时间后到达，从而创建延迟数据。这就是右侧示例中发生的事情：水位线在观察到该窗口的所有输入数据之前就超过了第一个窗口的结束，导致输出值不正确，而是5而不是14。这个缺点严格来说是启发式水位线的问题；它们的启发式性质意味着它们有时会出错。因此，仅依靠它们来确定何时实现输出是不足的，如果您关心正确性。

在第1章中，我对大多数需要处理无界数据流的强有力的乱序处理用例的完整性概念提出了一些非常强烈的声明。这两个缺点——水位线太慢或太快——是这些论点的基础。您只能从依赖于完整性概念的系统中获得低延迟或正确性的最佳效果。因此，对于那些想要同时拥有最佳效果的情况，一个人该怎么办？如果重复的更新触发器提供低延迟更新但没有关于完整性的推理方式，而水位线提供关于完整性的概念但具有可变且可能具有高延迟的特性，那么为什么不将它们的力量结合起来呢？

何时触发：早期/准时/延迟触发 FTW！#

我们现在已经看过了两种主要类型的触发器：重复更新触发器和完整性/水位线触发器。在许多情况下，它们单独使用都不足够，但将它们结合在一起就可以了。Beam通过提供标准水位线触发器的扩展来识别这一点，该扩展还支持水位线两侧的重复更新触发。这被称为早期/准时/延迟触发器，因为它将由复合触发器实现的窗格分为三类：

  • 零个或多个早期窗格，这是由重复更新触发器产生的结果，该触发器会定期触发，直到水位线通过窗口的末端为止。这些触发生成的窗格包含推测结果，但允许我们观察随着新的输入数据到达窗口而窗口的演变。这弥补了水位线有时过于缓慢的缺点。
  • 一个准时窗格，这是由完整性/水位线触发器在水位线通过窗口的末端之后触发的结果。这个触发很特别，因为它提供了系统现在相信该窗口的输入是完整的断言。这意味着现在可以推断缺少的数据，例如在执行外部连接时发出部分连接。
  • 零个或多个延迟窗格，这是由另一个（可能不同的）重复更新触发器产生的结果，该触发器会在水位线通过窗口的末端之后定期触发任何延迟数据。对于完美的水位线，将始终没有延迟窗格。但是，在启发式水位线的情况下，水位线未能正确计算的任何数据都将导致延迟触发。这弥补了水位线过于快的缺点。

让我们看看这在实际中是什么样子。我们将更新我们的管道，使用具有一分钟对齐延迟的周期处理时间触发器进行早期触发，以及针对延迟触发的每个记录触发器。这样，早期触发将为我们的高容量窗口提供一定量的批处理（由于触发器每分钟只触发一次，无论窗口的吞吐量如何），但我们不会为延迟触发引入不必要的延迟，如果我们使用一个相对准确的启发式水位线，那么延迟触发应该是相对较少的。在Beam中，它看起来像Example 2-7（Figure 2-11显示结果）。

Example 2-7.通过早期/准时/延迟API进行早期、准时和延迟触发

1
2
3
4
5
6
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());

Figure 2-11.带有早期、准时和延迟触发的流式引擎的窗口求和

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

  • 对于第二个窗口中“水位线过慢”的情况[12:02，12:04）：我们现在每分钟提供定期的早期更新。最大的差异在完美的水位线情况下，从几乎七分钟的时间到第一次输出减少到三分半钟；但在启发式情况下，它也显然得到了改善。两个版本现在都提供随着时间的推移而稳定的改进（带有值为7、10、18的窗格），在输入变为完整和窗格的最终输出材料化之间的延迟相对较小。
  • 对于第一个窗口中“启发式水位线过快”的情况，[12:00，12:02）：当值为9的数据出现延迟时，我们立即将其合并到新的校正窗格中，其值为14。

这些新触发器的一个有趣的副作用是它们有效地将完美水位线和启发式水位线版本之间的输出模式规范化。在Figure 2-10中的两个版本有着鲜明的不同之处，而在这里的两个版本看起来非常相似。它们看起来也更类似于Figures 2-6到2-8中的各种重复更新版本，但有一个重要的区别：由于使用水位线触发器，我们还可以在使用早期/准时/延迟触发器生成的结果中推断输入的完整性，这使我们可以更好地处理关心缺少的数据的用例，例如外部连接、异常检测等。

此时完美和启发式早期/准时/延迟版本之间最大的差异是窗口生命周期边界。在完美水位线的情况下，我们知道在水位线通过窗口的末端之后将不会再看到窗口的任何数据，因此我们可以在那个时间丢弃窗口的所有状态。在启发式水位线的情况下，我们仍然需要保留窗口的状态一段时间来解决延迟数据的问题。但是，到目前为止，我们的系统没有任何好的方法来知道需要为每个窗口保留状态的时间有多长。这就是允许延迟的作用所在。

何时: 允许延迟（垃圾回收）#

在我们进入最后一个问题（“如何进行结果的改进？”）之前，我想谈一下长期、无序流处理系统中的一个实际必要性：垃圾回收。在图2-11中的启发式水位线示例中，每个窗口的持久状态在整个示例的生命周期内一直存在；这是必要的，以便在数据到达时/如果到达时可以适当地处理它们。但是，虽然保留所有持久状态直到永远可能很好，但实际上，在处理无界数据源时，通常不实用保留给定窗口的状态（包括元数据）无限期；我们最终会用尽磁盘空间（或至少会厌倦为其付款，因为较旧数据的价值随时间降低）。

因此，任何现实世界的无序处理系统都需要提供一种方法来限制它正在处理的窗口的生命周期。一个干净而简洁的方法是通过定义系统内允许延迟的时间范围来实现；也就是说，为系统在处理时要费心处理的任何给定记录设置一个延迟限制；超出此时间范围的任何数据都会被简单地丢弃。在您限制了单个数据可以被延迟的时间之后，您也确定了窗口状态必须保留多长时间：直到水位线超过窗口结束时的延迟时间范围。但是，此外，您还赋予了系统立即丢弃任何超过延迟限制的数据的自由，这意味着系统不会浪费资源处理任何人都不关心的数据。

1
2
3
4
5
6
7
                            **测量延迟**

使用导致第一次出现迟到数据的度量标准（即启发式水位线）来指定处理迟到数据的时间范围可能有些奇怪。在某种意义上，它确实是这样的。但是在可用选项中，它可能是最好的选择。唯一的其他实际选项将是按处理时间指定时间范围（例如，水位线通过窗口结束后的10分钟保留窗口），但是使用处理时间将使垃圾回收策略容易受到管道本身的问题（例如，工作人员崩溃，导致管道停滞几分钟），这可能会导致实际上没有机会处理迟到数据的窗口。通过在事件时间域中指定时间范围，垃圾回收与管道的实际进度直接相关联，从而降低了窗口错过适当处理迟到数据的机会的可能性。

但是请注意，并非所有水位线都是相同的。当我们在本书中谈论水位线时，通常是指*低*水位线，它们悲观地尝试捕获系统已知的*最旧*未处理记录的事件时间。使用低水位线处理迟到数据的好处是它们对事件时间偏差的变化具有弹性；无论管道中的偏差增长多大，低水位线始终会跟踪系统已知的最旧未完成事件，提供可能的正确性保证。

相反，某些系统可能会将“水位线”一词用于其他目的。例如，Spark Structured Streaming中的水位线是*高*水位线，它们乐观地跟踪系统已知的*最新*记录的事件时间。在处理迟到数据时，系统可以自由地丢弃早于高水位线调整了某些用户指定的延迟阈值的任何旧窗口。换句话说，系统允许您指定预计在管道中看到的最大事件时间偏差，然后丢弃任何超出该偏差窗口的数据。如果管道内的偏差保持在某个恒定的增量内，这可能很有效，但比低水位线方案更容易错误地丢弃数据。

由于允许延迟和水位线之间的交互有些微妙，因此很值得看一下示例。让我们把示例2-7/图2-11中的启发式水位线管道添加一个1分钟的延迟时间范围（请注意，这个特定延迟时间范围之所以被选择是严格因为它能很好地适合图表；对于实际用例，一个较大的延迟时间范围可能会更实用）：

示例2-8。允许迟到的早期/按时/晚期点燃

1
2
3
4
5
6
7
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的完美水位线部分中看到的。
  • 即使在使用启发式水位线时，需要指定延迟时间范围，也有一个值得注意的例外情况，那就是对于针对有限数量的键的所有时间的全局聚合（例如，按网页浏览器系列分组计算您网站上的总访问次数）。在这种情况下，系统中活动窗口的数量受使用的有限键空间的限制。只要键的数量保持在可管理的低水平，就不需要担心通过允许延迟来限制窗口的生命周期。

现在，让我们满足实际需求，继续第4个也是最后一个问题。

方法：累加#

当使用触发器在单个窗口上生成多个窗格时，我们面临的最后一个问题是：“如何关联结果的细化？在我们迄今看到的示例中，每个连续的窗格都建立在紧接着的窗格之上。但是，实际上有三种不同的累加模式：

  • 丢弃

    每次实现窗格时，任何存储的状态都被丢弃。这意味着每个连续的窗格都独立于之前的任何窗格。当下游消费者执行某种聚合时，例如将整数发送到一个希望接收将它们相加以生成最终计数的增量的系统时，丢弃模式是有用的。

  • 累加

    与图2-6到2-11中的情况一样，每次实现窗格时，任何存储的状态都被保留，并且将来的输入被累加到现有状态中。这意味着每个连续的窗格都建立在以前的窗格之上。当稍后的结果可以简单地覆盖以前的结果时，例如在像HBase或Bigtable这样的键/值存储中存储输出时，累加模式是有用的。

  • 累加和撤回

    这就像累加模式，但在生成新窗格时，它还会为先前的窗格生成独立的撤回。撤回（与新的累积结果相结合）本质上是明确地说“我先前告诉过你结果是X，但我错了。去掉上次告诉你的X，并用Y替换它。”有两种情况特别有用：

    • 当下游消费者通过不同的维度重新分组数据时，新值可能会与以前的值有不同的键，并因此进入不同的组。在这种情况下，新值不能仅覆盖旧值；您需要撤回以删除旧值。
    • 当使用动态窗口（例如会话，在接下来的几分钟中我们将更详细地了解）时，由于窗口合并，新值可能会替换多个先前的窗口。在这种情况下，仅从新窗口中确定被替换的旧窗口是困难的。为旧窗口提供明确的撤回使任务变得简单。我们在第8章中详细介绍了这个例子。

每个组的不同语义在并排看起来会更加清晰。考虑图2-11中第二个窗口（事件时间范围为[12:06,12:08)）的两个窗格。表2-1显示了每个窗格在三种累加模式（其中累加模式是图2-11本身使用的特定模式）中的值是什么。

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

丢弃 累加 累加和撤回
窗格1：输入=[3] 3 3 3
窗格2：输入=[8, 1] 9 12 12,–3
最终正常窗格的值 9 12 12
所有窗格的总和 12 15 12

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

  • 丢弃

    每个窗格仅包含在该特定窗格期间到达的值。因此，观察到的最终值并未完全捕获总和。但是，如果您将所有独立的窗格本身相加，您将得到正确的答案12。这就是为什么当下游消费者本身执行材料化窗格的某种聚合时，丢弃模式很有用的原因。

  • 累加

    与图2-11相同，每个窗格包含该特定窗格到达的值以及以前窗格的所有值。因此，观察到的最终值正确地捕获了总和12。但是，如果您总结单独的窗格本身，您将有效地重复计算来自第1个窗格的输入，从而给出错误的总和15。这就是为什么当您可以简单地使用新值覆盖先前的值时，累加模式最有用：新值已经包含了迄今为止看到的所有数据。

  • 累加和撤回

    每个窗格都包括一个新的累加模式值和上一个窗格的撤回。因此，最后一个观察到的值（不包括撤回）以及所有材料化窗格（包括撤回）的总和都为12，这就是为什么撤回如此强大的原因。

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

示例2-9.早期/按时/迟到触发的丢弃模式版本

1
2
3
4
5
6
7
8
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-10.早期/按时/迟到触发的累加和撤回模式版本

1
2
3
4
5
6
7
8
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-9、2-11（仅启发式）和并排的最终帧，提供了三种模式的良好视觉对比。

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

正如您可以想象的那样，按照呈现的顺序的模式（丢弃，累加，累加和撤回）在存储和计算成本方面都是越来越昂贵的。为此，累加模式的选择为沿着正确性、延迟和成本的轴进行权衡提供了另一个维度。

摘要#

通过本章的学习，您现在已经了解了鲁棒流处理的基础知识，已经准备好在世界上做出惊人的事情了。当然，还有八个更多的章节等待您的关注，所以希望您不会在此刻非常急切地前进。但无论如何，让我们回顾一下我们刚刚讨论的内容，以防止您在匆忙前进时忘记了它。首先，我们触及的主要概念：

  • 事件时间与处理时间

    重要的区别是事件发生的时间和数据处理系统观察到它们的时间。

  • 窗口化

    通过沿着时间边界（在处理时间或事件时间中）切片管理无界数据的常用方法，尽管在Beam模型中我们将窗口化的定义限制为仅在事件时间内。

  • 触发器

    用于指定在特定用例中何时对输出进行实体化的声明性机制。

  • 水位线

    在事件时间中提供进度的强大概念，为在无界数据上运行的乱序处理系统中推理完整性（因此缺少数据）提供手段。

  • 积累

    为单个窗口的结果的细化之间的关系，对于多次演化时它的实体化情况。

  • 第二，我们用来构建我们的探索的四个问题：

    • 计算什么结果？= 变换。
    • 在事件时间中的哪个位置计算结果？= 窗口化。
    • 结果在处理时间中的何时实体化？= 触发器加水位线。
    • 结果的细化如何关联？= 积累。

第三，为了凸显这种流处理模型所提供的灵活性（因为最终，这就是所有事情的平衡：正确性、延迟和成本等竞争张力），我们仅仅通过最少的代码更改就能在相同的数据集上实现各种主要输出的回顾：

整数求和

示例2-1 / 图2-3

整数求和

固定窗口批处理

示例2-2 / 图2-5

整数求和

固定窗口流处理

每个记录触发器的重复

示例2-3 / 图2-6

整数求和

固定窗口流处理

整数求和

固定窗口流处理

整数求和

固定窗口流处理

重复对齐延迟触发器

示例2-4 / 图2-7

重复的

未对齐延迟

触发器

示例2-5 / 图2-8

固定窗口流式处理

启发式水位线触发器

示例2-6 / 图2-10

整数求和

固定窗口流处理

早期/准时/延迟触发器

丢弃

示例2-9 / 图2-13

整数求和

固定窗口流处理

早期/准时/延迟触发器

积累

示例2-7 / 图2-11

整数求和

固定窗口流处理

早期/准时/延迟触发器

累积和撤回

示例2-10 / ???

话虽如此，到目前为止，我们只看了一种窗口化类型：事件时间的固定窗口化。正如我们所知，窗口化有许多维度，我想在我们结束Beam模型之前至少再涉及其中两个。不过，首先，我们将稍微偏离主题，深入探讨水位线的世界，因为这些知识将有助于框架未来的讨论（并且本身也很有趣）。请欢迎Slava，右台……

  1. 如果你足够幸运能读到 Safari 版本的书籍，你会看到完整的类似“流式媒体 102”中的时间-lapse 动画。对于印刷、Kindle 和其他电子书版本，这些都是静态图像，带有到网络动画版本的链接。
  2. 请耐心等待。复合标点符号（即表情符号）的细粒度情感表达在 O’Reilly 出版物中是严格禁止的 。
  3. 的确，我们在 Beam 的原始触发器功能中就是这样做的。回想起来，我们有点过头了。未来的迭代将更简单、更易于使用，在本书中我只关注那些可能以某种形式保留的部分。
  4. 更准确地说，函数的输入实际上是上游管道中水位线被观察到的点上游的所有内容在时间 P 的状态:输入源、缓冲数据、正在处理的数据等等;但从概念上来讲，把它看作是从处理时间到事件时间的映射更简单。
  5. 请注意，我选择明确省略启发式水位线的 9 值，因为它将帮助我就延迟数据和水位线滞后提出一些重要观点。实际上，一个启发式水位线可能会同样有可能省略其他某些值，这反过来可能会对水位线产生显著不那么严重的影响。如果你的目标是从水位线中筛选迟到的数据（在某些情况下非常有效，例如在滥用检测中，你只想尽快看到大多数数据），你不一定要选择启发式水位线而不是完美水位线。你真正想要的是百分位水位线，它明确地从其计算中删除一些迟到的数据百分位。见第三章。
  6. 这并不是说没有那些主要关心正确性而不是延迟的使用情况；在这些情况下，使用准确的水位线作为管道输出的唯一驱动器是一种合理的方法。
  7. 正如我们之前所知道的，这个断言要么是在使用完美水位线的情况下保证的，要么是在使用启发式水位线的情况下的一个经过教育的猜测。
  8. 你可能会注意到应该有第四种模式：丢弃和撤销。在大多数情况下，这种模式并不是非常有用，所以我在这里不再讨论它。
  9. 回想起来，选择一组更加面向材料化流中数据的观察性名称（例如“输出模式”）而不是描述产生这些数据的状态管理语义的名称可能会更清晰。例如：丢弃模式 → Δ 模式，累积模式 → 值模式，累积和撤回模式 → 值和撤回模式？然而，丢弃/累积/累积和撤回名称已经在 Beam Model 的 1.x 和 2.x 衍生版本中得到了确认，因此我不想通过偏离来引入潜在的困惑。此外，随着 Beam 3.0 的推出和汇合触发器的引入，累积模式很可能会更加淡化。在第 8 章中讨论 SQL 时会更详细地讨论这个问题。

Draft Here#