带你认识流计算（上篇） | 会员专栏

本文介绍了流计算的概念和技术要素，简单比较了三种主流的流计算框架Structured Streaming（Spark）、Flink和Kafka Streams。第1、2两节的目的是希望读者清晰理解流计算的一些重要概念和技术要点，尤其是其中一些容易混淆的地方，例如流计算与实时计算的关系、窗口化及水位线（watermark）和一致性模型等。第3节比较了一些主流计算框架，旨在为那些需要做系统选型的读者提供一些参考。

阅读本文不要求读者对流计算或其它大数据技术已经有一定基础。但在阅读第3节关于流计算框架比较之前，如果对Hadoop和Spark有一些经验，能更好理解架构上的讨论。

“流”一般是指源源不断的数据（Unbounded Data）。在理解流计算之前，需要先理解其对立概念——批计算（Batch Computing）。批计算假定在计算发生之前已经获得待处理的全部数据，因此可以一次处理所有的数据。相对应的，流计算是指发生在源源不断的数据之上的计算，因此在计算发生时刻数据并未完全抵达，甚至尚未产生。流计算的过程就像工厂里的流水线一样——一件产品在传送带上经历多道工序，其中每一道工序都处理从上游源源不断输送来的加工件，处理完成后再向下游输送。流计算与流水线有许多相似性，例如流水线一般都有多道加工工序，对应流计算过程中的多个算子；工厂里一般有多条流水线并行加工，对应流计算的并行计算；在流水线的某一道工序上，可能需要等接收到上游一批加工件之后才开始加工，对应流计算里的窗口化；等等。

按流计算的定义，流计算只与处理数据是否有限（即bounded还是unbounded）有关。但人们提到流计算，往往会联想到一些属性，比如“实时”、“低延迟”等等。实际上流计算与这些属性之间并没有直接联系。例如我们每天定时收到天气预报，而气象数据显然是一种流数据。天气预报系统在每天播报前处理一段时间的气象数据，那么数据的处理延迟可达数小时。

那么为什么人们会产生“实时”或“低延迟”的联想呢？这主要受到早期实时计算引擎的影响。实时计算是面向低延迟场景产生的一类计算框架，例如早期的Storm（Storm的新版本也在逐渐丰富流计算的语义，如窗口化）：

实时计算与流计算的共同点在于两者都可以处理流数据，但区别在于两者的出发点不同。在成熟的流计算引擎出现之前，人们采用基于分批的模式处理流数据（类似下图），源源不断的数据按照抵达系统时间分批依次处理。

但在一些场景下这种方式的延迟过高，于是人们引入了实时计算来弥补批计算的不足。例如在运营社交网站时，我们可能既希望准确统计前一天的热点词汇（这采用批计算），又希望在当天实时发现一些动态情况（这采用实时计算）。实时计算最初设计的出发点就是降低计算延迟，在语义上无法兼顾批计算的场景，只能与批计算形成一种互补关系。

基于实时计算与批计算的互补关系，Nathan Marz提出了一种称为Lambda的系统架构，其本质是批计算与实时计算的一种混合架构，如下图所示：

在Lambda体系下，同一份数据会分别进入批计算子系统（batch sub-system）和实时计算子系统（realtime sub-system）。批计算子系统存储完整的数据，计算出完整的、精确的结果；实时计算子系统只处理最近的数据，计算出低延迟的、粗糙的结果。于是，一个Lambda系统兼顾结果的实时性和准确性。Lambda看起来非常完美，却存在一些问题。首先，运维两套子系统是比较繁琐的；此外，很多时候两个子系统要完成的数据处理逻辑可能是相同的，但却分别在两个不同的子系统，存在较高的同步代价。

早期的实时计算引擎在实现上存在一致性问题。例如，大多数早期的流计算引擎只支持At-most-once（在2.5节会进一步说明）的处理语义。这意味着一些数据可能根本没有被处理就丢弃了，而另一些数据可能被重复处理了，计算得到的结果可能根本不正确。

批计算和流计算之间并不存在本质上的差别。批计算是流计算的一种特例，流计算可以处理无限数据，当然也能处理有限数据。人们之所以接受Lambda这样的体系是受限于当时缺乏成熟的流计算引擎。随着人们对流计算的理解不断深刻和流计算引擎的不断完善和丰富，出现了一系列能同时兼顾批处理和流处理的计算引擎，如Flink、Structured Streaming、Samza、Kafka Streams等等；还出现了Beam这样的抽象的计算定义模型。在这些引擎中虽然有时候仍然区分批计算和流计算的API，但两者的底层实现往往没有本质区别，仅为了方便用户理解。本文后续讨论里提到的流计算引擎不再是在Lambda架构中的为了追求低延迟而放弃准确性的实时计算子系统，而特指这些新的支持批流结合的计算引擎。

在不同的流计算引擎中使用的词汇可能不一样，所以为了方便讨论，这里借用Beam的词汇来讨论。Beam是一种建立在流计算引擎之上的描述性模型，其抽象层次更高。当然，由于Beam是基于Google Cloud Dataflow提出的，自然带有其特色，但仍然不妨碍作为理解流计算概念的入口。

从本质上看，流计算引擎所做的工作就是：对于源源不断的数据，怎么计算？如何缓存数据？在什么时机计算？如何更新计算结果？不同的流计算引擎的API基本上都是为了描述上面的4个核心问题。在Beam里，把这些问题概括为What、Where、When、How：

本文没有讨论第一个问题，即“What”。这是因为各种流计算引擎都提供了较为完备的计算算子，大家可以阅读Flink 、Structured Streaming或Spark Stream关于算子的描述文档，而这些内容比较直观，容易理解。一个流计算应用往往表示成数据源节点、数据输出节点和一些算子构成的有向图，如下图所示：

在下面，我们介绍流计算中一些非常重要的技术要素。在3.1和3.2中将讨论流计算引擎中时间和状态两个重要概念；在3.3里将基于这两个重要概念，讨论算子状态在时间上的管理策略（即窗口化），解决上面的“Where”问题；在3.4节讨论计算的时机，解决上面提到的“When”和“How”两个问题；在3.5中我们将着重讨论流计算引擎的一致性定义，这对理解引擎对正确性的保证非常重要。

流计算引擎中的时间一般有两种情况：

上述两个概念比较容易理解。假设我们正在开发一套用于设备高温报警的系统，设备的温度会通过网络传输到流计算引擎，我们根据温度的情况决定是否向管理员发送告警信息。

假设某设备上的温度传感器检测到温度为105°，此时的设备系统时间为t0，该读数通过网络传输抵达流计算引擎后，在t1时刻开始处理该信息。那么，该事件的发生时间就是，而处理时间就是。 虽然我们总希望发生时间与处理时间相同，但现实中会由于网络通讯、资源调度、处理模式等等原因造成延迟，所以处理时间总是在发生时间之后，即t1-t0>0。

注意这里的讨论基于一个假设，即数据源时钟和流计算系统时钟都是准确的。但在一些实际场景下，发生时间和处理时间都可能跟实际时间（Wall time）不同，甚至可能出现t1-t0<0的情况。造成这种情况的原因可能是产生数据的数据源或流计算引擎没有可靠的时间同步服务。时间不同步会造成计算结果错误，所以系统设计者需要慎重考虑，不过这些额外工作不在流计算引擎的考虑范围。

一个流计算应用通常包括多个算子，在定义流计算应用状态之前需要先理解什么是算子的状态。下面给出了一些常见的算子状态的例子：

可见，算子的状态通常是指为了完成当前的计算而保存的一些基于历史数据产生的内容。我们以第一个例子展开说明。假设我们想在数据流中查找下面形状的模式：

这个模式是由多个数据点构成，所以算子不能仅根据当前处理的一条记录完成模式匹配，而是必须缓存一段历史数据。下图展示了不同时刻节点的状态，以及随着状态的变化得到的不同匹配结果。在t0时刻算子状态中保存的数据与模式不匹配，在t1时间如果后续抵达的数据使算子状态如上面的分支，那么此时达成匹配；而如果后续抵达的数据使算子状态如下面的分支，则仍然为不匹配。

需要注意的是，算子并非一定有状态（state-full），也可以无状态（state-less)。例如，我们希望根据数据里的某种属性进行过滤，而这种属性跟被处理的数据无关（即不会随时间发生变化），那么节点就不需要记录任何状态信息。所以，算子是否有状态仅取决于算子本身的处理逻辑。

下面我们定义流计算应用的状态。一般情况下，流计算应用的状态定义为在某个时刻其中每个算子的状态的集合。但是，这里隐含一个前提条件——各个算子的状态是按数据对齐的。下面是按数据对齐的一个示例：

假设流计算应用中存在两个算子o1和o2算子（暂时忽略数据源和输出），算子o2位于算子o1的下游，假设两个算子都是有状态的。如果o1在处理完数据r1和r2之后，状态记做s1_1；而算子o1处理完r1和r2之后输出的结果记做r1+2，算子o2在处理完之后，状态记做s2_1。那么应用的状态sapp={s1_1,s2_1}，即按照数据对齐的每个算子状态的集合。相应的，在算子o1和o2都处理完数据r3和r4之后，应用新的状态sapp={s1_2,s2_2}。 强制应用内算子之间按数据对齐是为了在状态恢复时可以从统一的起点开始。数据的对齐方式可以是数据分批，或者在数据中插入一些标记（称为Marker或Barrier），这些具体的策略在阅读完第四节对流计算框架的介绍后可以更好理解。

要支持应用在故障后不必完全重新计算，还需要对状态进行持久化。但持久化有一定的IO、存储开销，所以一般情况下不会在处理完每条记录后都进行持久化，而是按一定的周期，比如每5秒、或每处理1000条记录等等。在有些计算引擎中，为了减少状态持久化的开销，支持增量的更新策略，即每次持久化只更新与上次不同的部分。另外，由于状态有存储开销，所以在编写流计算算子时需要注意不要使算子状态无限增长，例如可以引入Time-To-Live（TTL）或某种自定义的清理机制。

处理模式要解决的是前面提到的“Where”问题，即在什么时间范围的数据上执行计算。如果在处理之前我们已经得到了所有的数据，那么可以一次性处理所有数据；而如果数据源源不断到来，情况会复杂一些，我们需要分一些具体情况来讨论。

• 时间无关（Time-agnostic）

有时候处理数据流并不需要考虑数据的发生时间或处理时间。最典型的一个场景就是“过滤”（Filter），比如我们希望在后续处理中只考虑满足某些条件的数据（例如温度大于50°），那么在筛选过程中，我们并不需要考虑数据产生或处理时间。注意，这里所谓的“时间无关”并不一定是指数据的时间维度不重要，而是指在处理数据时不需要考虑时间。另一个时间无关的例子，在一些近似学习算法中（如streaming K-means)，每次接收到新的数据后迭代得到新的模型。

• 窗口化（Windowing）

有些情况下在处理数据前，需要考虑数据的产生时间或处理时间。最简单的一种场景是按照处理时间定期执行计算，例如每隔5分钟，系统将过去5分钟内抵达的数据处理一遍。注意这时候一条数据抵达系统后可能不会立刻被处理，而是需要等待一段时间。时间有关的处理往往都可以描述成一种“窗口化”的形式，而每一个窗口代表从时间维度来看，哪一段数据是计算的对象，比如例子里提到的每隔5分钟处理过去5分钟内抵达的数据。

窗口一般有两个要素：

前面提到了两种时间概念：发生时间和处理时间。于是，窗口化也大都按照这两种时间来考虑。基于处理时间的窗口化比较容易理解，比如每隔n分钟，处理过去m分钟积累的数据。如果n=m，那么时间窗口之间是没有重叠（non-overlapping）的；而如果m>n，则连续两次处理的数据是有一定重叠（overlapping）的。

基于发生时间的窗口化要稍微复杂一些（如上图下部所示）。以设备高温报警为例，我们希望按照发生时间来统计机器每5分钟内的平均温度。那么可能的时间窗口为

数据是有延迟地、乱序地抵达的，所以在8:05时刻是无法得到8:00~8:05所有数据的。因为延迟总是存在的，所以一般计算要延后一些。那么需要延后多少呢？实际上，这个问题如果脱离真实场景是无法回答的。如果传感器读数可以从传感器通过网络直接发送到处理系统，那么这里的延迟可能只有数毫秒（在网络无故障的前提下）；但如果传感器无法联网，先缓存在机器上，需要人工每天用硬盘拷贝后传输到处理系统，那么延迟可能是数十小时。

除了上述两种典型窗口化方式以外，还可以有一些自定义的窗口定义方式。一个最典型的例子是分析人们登录一个网站的行为——从登录到登出之间存在一个session，而若需要按照session来进行计算就需要将某个用户ID的每个session定义为一个窗口。窗口化的形式看似多样，其本质是在定义被计算数据所处的时间范围，即解决Where的问题。这里在讨论窗口化时，假定待处理数据在时间上是连续的，即窗口一般是一段连续的时间——这并非一种硬性要求，流计算引擎通常会提供一些自定义的实现方式。

窗口化处理模式意味着算子一定是有状态的（即state-full），并且状态有有限的生命周期。首先，窗口化意味着必须缓存一定数据或计算中间结果，以便在窗口内数据完整抵达后执行计算，所以一定是有状态的。然后，在窗口化的情况下，状态总是与某个窗口绑定的，一旦窗口内数据完整抵达并完成计算，状态就不再有意义，所以随着窗口生命期的完结，状态的生命期也随之结束。但是需要注意的是，并非所有算子状态都是有生命期的，例如我们想统计所有接收到的设备温度数据里的最大值，这时候状态就不会有明确的生命期限。最后，有状态的算子并不一定存在窗口化，想想前面提到的迭代训练机器学习模型的例子。

• 窗口生命期管理

我们来考虑一个窗口的生命周期。理解了这个问题可以方便我们了解流计算引擎的算子状态管理。流计算引擎的内部状态一般存储在参与计算节点的内存或磁盘，意味着一定的存储开销。基于处理时间的窗口生命期管理很简单，因为这与被处理的数据流无关，流计算引擎只需要根据系统时间定期新建或销毁窗口。

基于发生时间的窗口生命期管理要复杂一些。沿用前面计算窗口内平均温度的例子，考虑8:00~8:05这个窗口。由于数据是陆续抵达的，为了计算平均温度就必须缓存两个值，即落在该窗口内的所有温度读数的个数和它们之和，最终可以计算得到平均数：

窗口的生命期内引擎需要保存这两个值（即算子的状态），直到计算得到温度平均数为止。那么这个窗口是什么时候创建的呢？是在引擎接收到第一个落在该时间区间内的值的时候。根据前面的讨论我们知道，这个时间的具体值是与应用场景相关的。如下图所示，假设在9:00时刻接收到落在8:00~8:05窗口内的第一条记录，则此刻创建窗口。随着后续数据抵达不断更新窗口状态。

那么窗口会在什么时刻关闭呢？关闭窗口意味着完成计算并清空状态，严格意义上讲需要等窗口内数据都抵达后才能关闭窗口，但这个问题同样与应用场景相关。前面假定我们接收到一个窗口内读数的时刻是在9:00，说明延迟约1小时，那么在9:05时刻我们可能认为所有在8:00~8:05分的数据都已经被接收了，于是可以计算平均温度并关闭该窗口；或者，我们出于保守的目的，认为需要等到9:10才能确认数据完全接收。注意，无论我们最终选择9:05还是9:10，这都只是我们根据经验人为设定的一个估计值，并不能确保在此之前所有窗口内的数据都已经到达。当然，如果某个实时系统能够给出延迟的上界，我们就可以按照上界给出一个绝对可靠的时间。所以，流计算引擎并不能保证计算结果与事实一致，因为确保所有数据在给定时间之前抵达并非计算引擎需要考虑的问题，计算引擎要做的是提供完备的描述语义。

为了描述这种关闭窗口的时间，流计算引擎引入了一个称为水位线（Watermark）的概念。水位线实际上是一个映射关系，即根据当前系统的状态估计数据抵达情况。在前面的例子里，如果我们根据当前时间是9:10判断在8:05之前所有数据都已经抵达，则映射的输入是“当前时间9:10”，输出是“8:05”。当然也可以用其它状态来推测水位线，比如如果我们接收到8:05时刻的温度读数，则认为8:05之前所有数据已经抵达。并不存在完美普适的估计水位线的方法，而需要根据应用场景设定最适合的估计方法。

水位线一旦超过某个时间窗口的最大时间，则可以计算出窗口结果并关闭窗口。窗口关闭后，引擎就可以释放该窗口所占用的存储空间。在前面的例子里，计算平均值只需要保存两个值，但在有些场景下窗口内可能需要保存原始数据，那样的话对存储的占用开销会比较大。所以，准确地估计水位线非常重要。通常情况下，窗口保存的时间越短则存储代价越小、计算结果越粗糙；反之，则存储代价越大、计算结果越精确。计算引擎需要开发者来平衡开销和精度。

在一些流计算引擎里，除了水位线外还会有最大延迟的概念，即在水位线已经超过窗口最大时间后还需要维持一段时间。在某些场景下可能会使窗口定义更加灵活，但从窗口的生命期角度来说，最大延迟也只是水位线的一部分。

对某个时间窗口而言，在水位线超过窗口最大时间以后就可以计算得到该窗口对应的值。但在此之前，有时候我们也希望提前处理已经落在窗口内的数据，以便提前获得一些估计结果，这时候就需要用到触发器（Trigger）的概念。触发器决定了窗口计算的时机，例如，可以在窗口初始化之后，每隔一段时间计算一次中间值，或每接收到一定数量的数据后计算一次中间值；或者在接收到特定数据时计算一次中间值。当然，一般的计算引擎都对每个窗口自动注册一个默认触发器——在窗口关闭时计算一次。下图展示了随着水位线超过窗口最大时间后计算温度平均值的例子。

最终每个窗口只会保留一个计算结果，所以每次计算都涉及到如何更新窗口结果的问题。一般有两种更新方式：替换和累积。其中替换是指后一次的计算结果替换前一次的结果；累积是指后一次的计算结果与前一次的计算结果经过某种运算（如加法）得到新的结果。在前面关于计算5分钟内平均温度的例子里，如果我们定义的触发方式是每分钟（处理时间）计算一次中间结果，那么对于窗口8:00~8:05，我们依次得到的结果是：

窗口的第一次计算是在9:01，每次计算中间结果，都可以根据目前窗口内的所有数据来计算，所以可以替换掉前面已有的中间结果。而窗口8:00~8:05最终得到的计算结果在9:06。通过上述计算过程，我们不仅可以得到最终结果，还能观察到中间结果的变化过程。累积的更新形式往往用于那些可以分段计算的情况，比如我们可以需要统计一段时间内接收到数据的量，就可以将后续统计结果叠加在之前的中间结果之上。

在前面关于Lambda结构的讨论中，曾经提到实时计算给人的印象是“不准确”，这种不准确的根源是一致性问题。一般来说，针对流数据的计算一致性可以分为三种类型：

要理解上面的三种语义，首先要清楚这里提到的“丢失”和“重复”特指在流计算系统内部发生的行为，而不是指数据本身存在的问题。比如，数据在到达流计算引擎之前可能已经丢了，或数据本身就包含一定的重复率，那这些问题并不在流计算引擎考虑之内。那么，为什么流计算系统内会发生数据丢失或重复呢？这是因为流计算系统通常包含多个节点（如下图示例），节点间通过网络通信，并且节点本身也可能发生故障，导致数据重复或丢失。