流式系统 - 第一章: Streaming 入门（三）

Data Processing Patterns 数据处理模式

我们已经有了足够的背景知识，可以开始研究有边界和无边界数据处理中常见的主流类型：批处理和流处理。（在此我将微批处理和流处理相互等价，因为两者之间的差异在数据处理模式层面上并不大）

Bounded Data 有边界数据

处理有边界数据在概念上是很简单的，而且大家可能都很熟悉。在图1-2中，我们从左边开始有一个充满熵的数据集。我们通过一些数据处理引擎（通常是批处理，如果是设计良好的流处理引擎也行）来运行它，比如MapReduce，最后在右边有一个新的、更有用的结构化数据集。

图1-2. 用一个经典的批处理引擎进行有边界数据处理。左边的有边界非结构化数据池通过数据处理引擎运行，在右边产生相应的结构化数据。

在这个方案中，实际计算的可以有各种变化，但总体模型相当简单。现在让我们来看看处理无边界数据的各种典型方法，从传统的批处理引擎使用的方法开始，最后是可以在为无边界数据设计的系统中采取的方法，比如大多数流处理或微批处理引擎。

Unbounded Data: Batch 无边界数据: 批处理

批量引擎，尽管没有明确地为无边界数据而设计，但自从批处理系统首次被设想以来，一直被用来处理无边界数据集。这种方法主要将无边界数据切成适合批处理的有边界数据集的集合。

Fixed windows 固定窗口

使用批处理引擎的重复运行来处理无边界数据集，最常见的方法是将输入数据窗口化为固定大小的窗口，然后将每个窗口作为一个单独的、有边界的数据源来处理（称为tumbling windows滚动窗口），如图1-3。特别是对于像日志这样的输入源，事件可以被写进目录和文件的层次结构中，这些目录和文件的名称包含了它们所对应的窗口编码。这种事情乍一看很简单，因为已经进行了基于时间的整理，提前把数据放到了适当的事件时间窗口中。

然而，在现实中，大多数系统仍然有一个完整性的问题需要处理（如果一些事件由于网络分区而在写入到日志的途中被延迟了怎么办？如果事件是在全球范围内收集的，必须先转移到一个公有的地方再处理怎么办？如果事件来自移动设备呢？），这意味着我们需要某些措施（例如，延迟处理，直到所有的事件都确定被收集，或者在数据晚到时重新处理特定窗口的整个批次）。

图1-3. 通过经典批处理引擎的临时固定窗口处理无边界数据。一个无边界的数据集被预先收集到有限的、固定大小的有边界数据窗口中，然后通过连续运行批处理引擎来处理。

Sessions 会话

使用批处理引擎将无边界数据处理成更复杂的窗口策略时(比如会话)，批处理引擎会出现很大的问题。会话通常被定义为(对于一个特定的用户)活跃期，终止于一个不活跃的间隙。使用批处理引擎来计算会话时，会话常常会被分割成不同的批次，如图1-4中的红色标记所示。我们可以通过增加批处理量来减少分割的数量，但代价是增加延迟。另一个选择是增加额外的逻辑来拼接之前运行的会话，但代价是进一步的复杂性。

图1-4。通过经典批处理引擎的临时固定窗口将无边界数据处理成会话。一个无边界数据集被预先收集到有限的、固定大小的有边界数据窗口中，然后通过连续运行经典批处理引擎被细分为动态会话窗口。

不管怎么说，使用经典的批处理引擎来计算会话不太理想。一个更好的方法是以流的方式建立会话，这一点我们将在后面讨论。

Unbounded Data: Streaming 无边界数据: 流处理

流式系统是为无边界数据而生。对于许多现实世界的分布式输入源，数据不仅无边界，而且具备以下特性：

• 在事件时间方面高度无序。如果用户要按照数据发生顺序分析，就需要在管道中进行某种基于时间的shuffle。
• 具有不定的事件时间偏差。用户不能预期在某个固定的时长Y内，看到给定事件时间X的大部分数据。

在处理具有这些特征的数据时，可以采取一些技巧。我一般将这些方法分为四组：时间无关、近似算法、基于处理时间的窗口和基于事件时间的窗口。

Time-agnostic 时间无关

时间无关的处理用于时间不相关的场景，即所有的逻辑都是数据驱动的。基本上现有的所有流系统都支持时间无关的场景。批处理系统也很适合对无边界数据源进行时间无关的处理，只需将无边界数据源切成任意的有边界数据集序列，并独立处理这些数据集。

Filtering 过滤

时间无关处理的一个非常基本的形式是过滤，图1-5中呈现了一个例子。想象一下，你正在处理网络流量日志，用户想过滤掉所有不是来自某个特定域名的流量。用户在每条记录到达时查看它是否属于感兴趣的域名，不属于就丢弃。这类操作只依赖于一个单一的元素，所以无边界无序的数据源与事件时间偏差不具备相关性。

图1-5. 过滤无边界的数据。一个不同类型的数据集合（从左到右流动）被过滤成一个包含单一类型的同质集合。

Inner joins 内连接

另一个与时间无关的例子是内连接，如图1-6所示。当连接两个无边界数据源时，如果用户只关心来自两源的元素的连接结果，那么业务逻辑中就没有时间因素。在看到来自第一个源的值时，可以简单地缓存到在持久化状态中；只有在另一个源的值到达时，才发出内联的记录。

图1-6. 在无边界数据上执行内连接。当观察到来自两个源的匹配元素时，才发生连接。

如果把这个例子中的内连接(inner join)换成外连接(outer join)，就引入了我们谈论过的数据完整性问题：在你看到连接的一边数据之后，你怎么知道另一边是否会到达？说实话，你不知道，所以需要引入一些超时的概念，这就引入了一个时间元素。这个时间元素本质上是一种窗口化的形式，我们稍后会更仔细地研究它。

Approximation algorithms 近似算法

第二大类方法是近似算法，如近似Top-N、流式k-means等。它们接受一个无边界的输入源，并输出数据。如果你仔细看输出的话，这些数据近似符合正确结果，如图1-7所示。近似算法的优点是，在设计上，它们开销低，并且是为无边界数据而设计。缺点是这类算法不多，而且往往很复杂，并且它们的近似性质影响了它们的效用。

1-7. 在无边界数据上计算近似值。数据通过一个复杂的算法运行，产生的输出数据看起来近似符合预期结果。

这些算法通常在设计中确实有一些时间元素（例如某种内置的衰减），而且通常是基于处理时间的。这对那些为其近似值提供某种可证明的错误界限的算法来说尤其重要。需要记住的是，如果这些误差界限是以数据按顺序到达为前提的，那么当给算法提供具有不同事件时间偏差的无序数据时，这些误差界限基本上没有意义。

Windowing 窗口化

剩下的两种无边界数据处理方法都是窗口化的变种。在深入探讨它们之间的差异之前，我应该明确说明我所说的Windowing窗口化是什么意思，因为我们在上一节只简单地提到了它。窗口化的概念是指把一个数据源（无论是无边界的还是有边界的），沿着时间边界切成有限的小块进行处理。图1-8显示了三种不同的窗口化模式。

图1-8. 窗口化策略。每个例子都显示了三个不同的键，展示了aligned windows对齐窗口（适用于所有数据）和unaligned windows非对齐窗口（适用于数据的一个子集）之间的区别。

• Fixed windows (aka tumbling windows) 固定窗口（又称滚动窗口） 我们在前面讨论过固定窗口。固定窗口将时间切成具有固定大小时间长度的片段。通常情况下（如图1-9所示），固定窗口的片段是统一应用于整个数据集的，这是一个aligned对齐窗口的例子。在某些情况下，最好对数据的不同子集（如每个key）进行相位转移，以使窗口的完成负荷(completion load)在时间上分布得更均匀。
• Sliding windows (aka hopping windows) 滑动窗口（又称跳动窗口） 作为固定窗口的泛化，滑动窗口由一个固定的长度和一个固定的周期来定义。如果周期小于长度，窗口就会重叠在一起。如果周期等于长度，就会得到固定窗口。如果周期大于长度，就会得一段时间内部分数据子集的抽样窗口。与固定窗口一样，滑动窗口通常是对齐的，但是在某些使用情况下它们可以不对齐来进行性能优化。
• Sessions 会话 作为动态窗口的一个例子，会话是由一系列的事件组成的，以大于某个超时的不活动间隙为终点。会话通常将一系列时间上相关的事件组合在一起来分析用户在一段时间内的行为。会话的长度不能被先验地定义；它们取决于所涉及的实际数据。它们也是不对齐窗口的典型例子，因为会话在不同的数据子集（例如，不同的用户）中几乎没有相同点。

窗口化在两个时间域（处理时间和事件时间）都是有意义的，让我们详细看看每个领域，看看它们有什么不同。因为基于处理时间的窗口化比较常见，我们就从这里开始。

Windowing by processing time 按处理时间窗口化

当按处理时间窗口化时，系统将传入的数据缓冲到窗口中，直到一定的处理时间过去。例如，在5分钟固定窗口的情况下，系统将缓冲数据5分钟处理时间，之后它将把在这5分钟内观察到的所有数据作为一个窗口，发送到下游进行处理。

图1-9. 按处理时间窗口化进入固定窗口。数据根据它们到达管道的顺序被收集到窗口。

按处理时间窗口化有几个不错的特性：

• 简单。实现起来非常简单，不用担心洗数据的问题。只需在数据到达时进行缓冲，并在窗口关闭时将它们发送到下游。
• 直白地判断窗口的完整性。因为系统清楚地知道一个窗口的所有输入是否被观察到，可以对一个给定的窗口是否完整做出判断。这意味着在通过处理时间进行窗口化时，不需要处理"迟到"的数据。
• 帮助用户推断出关于源的信息。许多监控场景都属于这个类别。比如跟踪每秒发送到全球规模的网络服务的请求数。通过计算这些请求的速率来检测故障，就是对处理时间窗口化的运用。

但是处理时间窗口化有一个非常大的缺点：如果这些数据与事件时间相关，那么这些数据必须按事件时间顺序到达，处理时间窗口才能反映真实发生的事件。不幸的是，在现实世界的分布式输入源中事件时间顺序的数据并不常见。

一个简单的例子，想象一下收集使用统计数据供以后处理的ap。对于手机可能在任何时间内离线的情况（短暂的掉线，或者进入飞行模式，等等），在此期间记录的数据将不会被上传，直到设备再次上线。这意味着，数据到达时可能有几分钟、几小时、几天、几周或更长时间的事件时间偏差。如果以处理时间为窗口，基本上不可能从这样的数据集中得出任何有用的推论。

再比如，当整个系统正常运行时，许多分布式输入源可能会提供事件时间有序（或非常接近）的数据。不幸的是，输入源正常所以事件时间偏移较低这种事情很难保证。想象一个处理在多个大洲收集来的数据的全球服务。跨洲线路出现网络问题降低了带宽或增加了延迟。一下子，一部分输入数据可能出现更大的到达偏移。如果用户按处理时间为这些数据设置窗口，窗口就不再代表实际发生的数据；相反，它们代表事件到达处理管道时的时间窗口，这造成旧数据和当前数据的随机混合。

在这两种情况下，用户真正想要的是以事件时间来窗口化数据，其解决方案是对事件的到达顺序具有鲁棒性。我们真正需要的是事件时间窗口化。

Windowing by event time 按事件时间窗口化

当用户需要用有限的、反映事件实际发生时间的区域来观察一个数据源时，就会使用事件时间窗口化。它是窗口化的黄金标准。在2016年之前，大多数的数据处理系统缺乏对它的原生支持（尽管任何具有一致性模型的系统，如Hadoop或Spark Streaming 1.x，可以作为构建这样一个窗口化系统的合理底层）。今天，从Flink到Spark到Storm到Apex，多数系统都原生支持某种事件时间窗口化。

图1-10显示了一个将无边界的源分成一小时固定窗口的例子。

图1-10. 按事件时间窗口化进入固定窗口。数据被收集到基于其发生时间的窗口中。黑色箭头标出了处理时间与事件时间窗口不同的数据。

图1-10中的黑色箭头标出了两个特别的数据片段。数据到达的处理时间窗口与所属的事件时间窗口不一致。因此，如果这些数据在一个关心事件时间的用例中被纳入处理时间窗口，那么计算出来的结果将是不正确的。事件时间的正确性是使用事件时间窗口的一个优势。

在无边界数据源上进行事件时间窗口化的另一个好处是，用户可以创建动态大小的窗口，比如会话，而不会出现在固定窗口上生成会话时观察到的随机分裂，如图1-11所示。

图1-11. 按事件时间窗口化进入会话窗口。数据被收集到会话窗口，根据相应事件发生的时间来捕捉活动。黑色的箭头指出了将数据放入正确的事件时间位置所需的时间整理(temporal shuffle)。

当然，强大的语义的代价是事件时间窗口有两个明显的缺点，这是因为窗口的生命周期往往必须比窗口本身的实际长度更长：

• Buffering 缓冲 为了延长窗口的生命周期，需要更多的数据缓冲。幸好，持久化存储通常是数据处理系统的资源类型中最便宜的。因此，在使用数据处理系统时，这个问题通常比较小。此外，许多有用的聚合函数不需要缓冲整个输入集（例如，sum或average），而是可以递进执行，在持久化状态中存储一个更小的中间聚合状态。
• Completeness 完整性 鉴于我们不知道获得某个窗口的所有数据的确切时间，那要怎么知道这个窗口的结果什么时候可以获得？事实上，我们根本不知道。对于许多类型的输入，系统可以通过类似于MillWheel、Cloud Dataflow和Flink中的watermark水印对窗口的完成性给出准确的启发式估计。但对于绝对正确性非常重要的情况（比如计费），唯一真正的选择是为管道创建者提供一种方法，以表达他们希望窗口的结果何时被实现，以及这些结果应如何随着时间的推移而被完善。