1.2　数据处理模式

此时，我们已经拥有了足够的背景知识，可以开始研究目前在有界数据处理和无界数据处理中常见的核心使用模式。我们基于人们关注的两种主要的引擎（批处理引擎和流式引擎，在这种情况下，我实质上将微批处理归为流式，因为这二者之间的差异在这个层面的讨论中并不十分重要）讨论以上两种类型的处理。

1.2.1　有界数据处理模式

处理有界数据在概念上非常简单，也被大家所熟知。在图1-2中，我们从左侧一个充满无序状态的数据集开始。我们通过一些数据处理引擎（通常是批处理引擎，尽管设计良好的流式引擎也可以完成同样的工作），例如MapReduce，来运行这个数据集，在右侧输出一个具有更大内在价值的新的结构化数据集。

当然，作为这个方案的一部分，实际上你可以进行的计算有无限变种，但是总体模型非常简单。更有趣的是处理无界数据集的任务。现在我们来了解一下通常用来处理无界数据的各种方式，先介绍传统的批处理引擎使用的方法，再介绍大多数流式引擎或微批处理引擎等为无界数据而设计的系统使用的方法。

1.2.2　无界数据处理模式：批处理

尽管批处理引擎并不是明确地为无界数据而设计的，但自从批处理系统第一次被构想出来，它就被用来处理无界数据集。正如人们所期望的，这种方法围绕着将无界数据切分成适合批处理的有界数据集的集合。

1．固定窗口

使用批处理引擎的重复运行来处理无界数据集的最常见方法是，将输入数据以固定大小的窗口进行开窗，然后将每个这样的窗口作为单独的有界数据源（有时也称为 滚动窗口 ）进行处理，如图1-3所示。特别是对于像日志这样的输入源，事件被写入不同的目录和文件中，这些目录和文件层次的名称进行了编码以便与它们的窗口进行对应。这样看来，似乎事情变得非常简单了，你只需要执行一个基于时间的混洗处理，就可以提前使数据进入适当的事件时间窗口中。

在实际使用中，大多数系统仍然需要处理数据的完整性问题（如果由于网络分区的原因，一些事件在发送到日志系统的途中被延迟了，应该如何处理？如果事件需要全局收集，并且必须在处理之前迁移到公共位置，应该如何处理？如果事件来自移动设备，应该如何处理？）。这就意味着，可能需要一些缓解这方面问题的方案（例如延迟处理以确保所有事件都被收集完成，或者当迟到数据到达时对某个给定窗口的整批数据重新处理）。

2．会话

当你尝试使用批处理引擎将无界数据划分到更复杂的开窗策略（如会话）时，这种方法更不适用。会话通常定义为被一段不活跃的间隔终止的一系列活跃周期（例如对于特定用户）。当使用典型的批处理引擎计算会话时，你经常会得到跨越不同批次被拆分的会话，如图1-4中的虚线标记所示。可以通过增大批规模的方法来减少拆分会话的数量，但这样做的代价是增加延时。另一种方式是增加额外的逻辑来缝合之前进行的会话，但这会增加系统设计的复杂度。

无论采用何种方式，用传统批处理引擎来计算会话的效果都不理想。更好的方式是用流式系统来建立会话，稍后我们会讨论。

1.2.3　无界数据处理模式：流式

与大多数基于批处理的无界数据处理方法的特定性质相反，流式系统天生就是为无界数据构建的。正如前面所讨论的，对于许多真实的分布式输入源，你会发现自己不仅在处理无界数据，还在处理具有以下特征的数据。

● 事件时间方面高度无序，这意味着，如果你需要在事件发生的上下文中分析数据，那么你需要在流水线中进行一些基于时间的混洗操作。

● 事件时间偏差不固定，也就是说，你不能假定自己可以在时间 Y 的某段固定偏差 ε 内看到给定的事件时间 X 的大部分数据。

在处理具有上述特征的数据时，可以采用现有的几种方法。我通常将这些方法分为4组：时间无关、近似算法、基于处理时间开窗和基于事件时间开窗。

我们现在将花一点儿时间来研究上述每种方法。

1．时间无关

时间无关处理用于与时间基本无关的场景，即所有相关逻辑都是由数据驱动的。因为这些使用场景的一切都是由更多数据的到达所决定的，所以除基本的数据交付之外，流式引擎实际上没有其他特别需要提供的支持。因此，所有现有的流式系统基本上都支持开箱即用的时间无关的使用场景（当然，你关心正确性，那么应该对一致性保证中的系统间的方差取模）。批处理系统也非常适合对无界数据源进行时间无关处理，只需简单地将无界数据源切割成有界数据集的任意序列，然后独立地处理这些数据集。我们将在本节中介绍几个具体的示例，但考虑到掌握时间无关处理比较简单（至少从时间角度来看），我不会在它上面花费更多的时间。

过滤。 时间无关处理的一种非常基本的形式就是过滤，如图1-5所示。假设你正在处理Web流量日志，希望过滤掉所有来自非特定域的流量。你会在每条记录到达时查看它，看它是否属于你感兴趣的域，如果不属于，则丢弃它。这类事情在任何时间只取决于一个元素，因此与数据源是否无界无序，以及是否存在不同的事件时间偏差都不相关。

内连接。 另一个时间无关的例子就是内连接，如图1-6所示。当连接两个无界数据源时，倘若你只关心来自两个源的元素到达时的连接结果，那么逻辑中就没有时间元素。一旦看到来自某一个源的一个值，你就可以直接把这个值缓冲在持久状态中。只有当来自另一个源的第二个值到达时，你才需要发出这条连接的记录。（事实上，你可能需要某种垃圾回收策略去清理未发出的部分连接，例如基于时间的回收策略。但是，对于几乎没有甚至完全没有未完成连接的使用场景，回收策略可能根本不是问题。）

将语义切换到某种外连接引入了我们已经讨论过的数据完整性问题：在观测到连接的一端的元素后，如何知道连接的另一端元素是否会到达？实话告诉你，你不会知道，因此，必须引入超时的概念，即引入一个时间元素。时间元素本质上就是开窗的一种形式，我们将在稍后进行更深入的探讨。

2．近似算法

第二大类方法是近似算法，如近似Top- N 算法和流式 k 均值算法等。这些近似算法将无界数据源作为输入，输出基本上满足我们预期的结果，如图1-7所示。近似算法的优点是，设计上开销很低，而且本身就是为无界数据而设计的；缺点是，这类算法数量有限，算法本身通常也很复杂（这使得很难演化出新的算法），而且它们的近似特性限制了它们的效用。

值得注意的是，这些算法在设计上通常都引入了时间元素（如某种内置衰减）。算法通常在元素到达系统时处理它们，因此时间元素通常都基于处理时间。这对在其近似值上提供某种可证明误差界的算法而言尤其重要。如果这些误差界是基于按顺序到达的数据预测的，那么当算法使用事件时间偏差变化的无序数据时，这些误差界基本上没有任何意义。我们需要记住这一点。

近似算法本身是一个引人入胜的主题，但由于它在本质上是时间无关处理（对算法本身的时间特征取模）的另一个例子，因此使用起来非常简单，鉴于目前关注的重点，我们不再进一步研究它。

3．开窗

剩下的两类无界数据处理的方法都是开窗的变体。在深入研究它们之间的差异之前，我先清晰地说明我所说的开窗的确切含义，因为我们只是在1.2.2节中简要提到过它。简单地从概念上讲，开窗就是将数据源（无界或有界）按时间切割成有限的数据块，然后进行处理。图1-8展示了3种不同的开窗策略。

我们来仔细观察每种开窗策略。

固定窗口（又称为滚动窗口）

我们之前讨论过固定窗口。固定窗口把时间按照固定的时间长度切分成片段（segment）。通常（如图1-9所示），固定窗口的片段会统一地应用于整个数据集，这是 对齐的窗口 的一个例子。在某些情况下，需要对数据的不同子集（如每个键）的窗口进行相移，以便窗口的完成负载能够更均衡地随时间分布，这是 未对齐的窗口 的例子，因为它随着数据状况的变化而变化。

滑动窗口（又称为跳跃窗口）

滑动窗口属于广义上的固定窗口，滑动窗口是根据固定的窗口长度以及固定的滑动周期定义的。如果滑动周期小于窗口长度，窗口会出现重叠；如果滑动周期等于窗口长度，就是狭义的固定窗口；如果滑动周期大于窗口长度，就会得到一个比较奇特的“采样窗口”，该窗口只会查询时间维度上的数据子集。与固定窗口一样，滑动窗口通常是对齐的，但在某些使用场景中，滑动窗口也可以出于性能优化的目的而不对齐。注意，图1-8中的滑动窗口是以体现滑动的感觉为目的而绘制的，实际上，所有5个窗口都将应用于整个数据集。

会话

会话是动态窗口的一个例子，会话由一系列事件组成，这些事件由大于某个超时的不活跃间隔终止。会话将一系列与时间相关的事件（如在一次会议中观看的一系列视频）分组在一起，通常用于分析随时间变化的用户行为。会话很有趣，因为它们的长度不能预先定义，它们依赖于所涉及的实际数据。会话是典型的未对齐的窗口的示例，因为在数据的不同子集（如不同的用户）中，会话几乎从不相同。

我们之前讨论的两个时间域（处理时间和事件时间）是我们真正关心的。 开窗在这两个时间域中都是有意义的，所以我们将详细了解这两个时间域，并了解它们的区别。由于处理时间开窗在历史上更为常见，因此我们将从处理时间开窗开始。

基于处理时间开窗。 当基于处理时间进行开窗时，系统本质上是在经过一些处理时间后，将传入的数据缓冲到对应的窗口中。例如，在以5 min为窗口周期的固定窗口策略中，系统会对5 min处理时间内的数据进行缓冲，然后将每5 min处理时间内观测到的所有数据视为一个窗口，并将这些窗口发送至下游进行处理。

基于处理时间开窗有几个很好的特性。

● 简单。基于处理时间开窗的实现非常简单，因为你不需要关注在时间范围内混洗数据的问题。你只需要在数据到达时将数据进行缓冲，然后在窗口结束时将数据发送至下游。

● 判断窗口完整性很简单。由于系统十分清晰地了解窗口的所有输入是否都已经可见，因此对于一个给定的窗口是否完整能够做出准确的判断。这意味着在基于处理时间开窗时，不需要具备任何处理“迟到”的数据的能力。

● 如果你想 在观测数据时 推断出数据源的信息，基于处理时间开窗恰好合适。许多监控类的场景都属于此类情况。假设需要跟踪每秒发送到服务全球规模的Web服务的请求数，以检测服务是否正常为目的而计算请求率是基于处理时间开窗的完美应用。

除了这些优点，基于处理时间开窗存在一个非常大的缺点： 如果所讨论的数据有与其相关的事件时间，而且基于处理时间的窗口反映这些事件实际发生的时间，那么这些数据就必须按事件时间的顺序到达。 遗憾的是，以事件时间排序的数据在许多真实的分布式输入源中并不常见。

举一个简单的例子，假设存在一个收集用户统计信息以供后续处理的手机应用。当给定的移动设备在任何一段时间内处于离线状态（短暂的连接中断、跨境飞行中启用飞行模式等）时，这段时间内记录的数据无法进行上传，直到该设备再次处于在线状态。这意味着，数据到达的时间可能偏离事件时间几分钟、几小时、几天、几周甚至更长。在基于处理时间开窗时，期望从包含这类数据到达时间和事件时间偏差过大的数据的数据集中提取任何有用的推断，基本上是不可能的。

另一个例子是，当整个系统正常运行时，许多分布式输入源似乎可以提供按照事件时间排序（或非常接近）的数据。当系统正常时，事件时间与输入源的偏差较低，但这并不意味着可以一直保持这种状态。假设存在一个全球性服务，处理从多个大洲收集的数据。如果带宽受限的洲际线路中发生的网络问题（这种情况非常普遍）进一步降低了带宽或者增加了延时，那么可能突然有一部分输入数据以比之前大得多的时间偏差到达。如果对这些数据基于处理时间开窗，那么这些窗口就不能像之前一样体现窗口中实际发生的数据的情况。相反，它们表示的是事件到达处理流水线时间的窗口，这条流水线由新旧数据随意混合而成。

在上述两个例子中，我们真正想要的是以保证事件到达的顺序的方式，基于事件时间对数据进行开窗。我们真正想要的是事件时间开窗。

基于事件时间开窗。 当需要以将反映事件实际发生的时间切成的有限数据块的方式观测一个数据源时，你需要使用事件时间开窗。这是开窗的标准范本。在2016年之前，大多数在用的数据处理系统缺乏对事件时间开窗的原生支持（尽管任何具有良好一致性模型的系统，如Hadoop或Spark Streaming 1.x，都可以作为构建此类开窗系统的合理基础）。我很高兴地看到，今天的世界看起来已有诸多改观，无论是Flink，还是Spark，或是Storm和Apex，这些系统本身都支持某种类型的事件时间开窗。

图1-10展示了一个将无界数据源开窗为1小时固定窗口的示例。

图1-10中的箭头指明了两部分特别有意义的数据，这两部分数据到达与数据所属的事件时间窗口不匹配的处理时间窗口。因此，对于关注事件时间的使用场景，如果将数据基于处理时间开窗，计算结果是不正确的。正如我们所期望的，事件时间正确性是使用事件时间窗口的一个好处。

在无界数据源上基于事件时间开窗的另一个好处是，可以创建窗口长度是动态的窗口，如会话，在固定窗口上生成会话时无须将数据进行任意的拆分（如在1.2.3节的会话示例中看到的那样），如图1-11所示。

当然，如此强大的语义不会免费提供，事件时间窗口也不例外。因为窗口的存在时间（在处理时间中）通常必须比窗口本身的实际长度长，所以事件时间窗口有以下两个明显的缺点。

缓冲

由于窗口生命周期的延长，系统需要进行更多的数据缓冲操作。值得庆幸的是，持久存储通常是大多数数据处理系统依赖的最便宜的资源类型（其他类型主要是CPU、网络带宽和RAM）。因此，当使用经过良好设计的具有强一致持久状态和良好的内存中的缓存层的数据处理系统时，这个问题通常没有想象的那样令人担忧。此外，很多有用的聚合操作不要求对整个输入集进行缓冲，取而代之的是，使用存储在持久状态中的更小的中间聚合进行增量计算。

完整性

考虑到我们没有好的方式来判断什么时候一个给定窗口的数据已经全部到齐，那么如何才能知道窗口的结果何时准备物化输出呢？实际上，我们没办法知道。对于许多类型的输入，系统能够通过某种在MillWheel、Cloud Dataflow和Flink中可以找到的类似于水位（在第3章和第4章中会详细讨论）的概念，对窗口的结束时间进行大致准确的启发式的估计。但是，当把保证结果绝对正确摆在首位时（例如在计费系统场景中），唯一的选择是为流水线的构建者提供一种方式来表达何时物化窗口的结果，以及如何随着时间的变化来改进这些结果。处理窗口完整性（或应对缺少完整性）是一个很有吸引力的话题，但最好在具体示例的上下文中探讨，我们将在后面介绍。 Se1N7L4unXIrM3MbQW5lSWTvP18pua5MeUWqrxj7nveac9C6kOW5kvo28Zhtr5uj