初识流计算

流计算是最近几年发展较快的一项数据处理技术。2011年Storm开源后的广泛应用使得流计算开始得到广泛关注，随后出现的Spark Streaming、Flink等流计算引擎进一步满足了应用在更多场景下的需求，流计算也获得更加蓬勃的发展。那么什么是流计算呢？已经有批计算了，为什么还需要流计算？流计算又是如何处理数据的？业界对流计算并没有一个统一的定义，对流数据的处理方式也存在不少差异。但其中，数Google在2015年发表的论文《The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive Scale, Unbounded, Out of Order Data Processing》所提出的概念和处理模型最被广为接受。这篇论文的观点后来得到进一步完善并在其作者出版的《Streaming Systems》一书中阐述。本文结合书中观点和自己的理解，为读者呈现流计算的What、Why和How，让读者对流计算有个初步的认识。

作者：苏毓仁
来源：建信金科大数据

What：什么是流计算？

在介绍流计算的概念前，我们先来看看什么是流。我们将数据根据数据集的大小分为两类：

• 有边界数据（Bounded Data）：数据集的大小是有限的，它有一个开始和一个结束的边界。
• 无边界数据（Unbounded Data）：数据集的大小在理论上是无限的，它有一个开始的边界，但没有结束的边界，它会随着时间的增长而不断产生新的数据。
  

流（也称流数据）就是对无边界数据集的抽象表示。流计算（也称流式处理）就是针对无边界数据的数据处理，而流计算引擎则是针对无边界数据而设计的一类数据处理引擎。 我们所熟知的批计算，针对的是有边界数据的处理，这些数据一般是预先准备好的，并且在某些既定的时间点去触发数据的处理，比如信用卡系统可能在每天凌晨对当天账单日的用户过去一个月的交易记录进行汇总统计，这过去一个月的交易记录就是一个有边界的数据集，完成这个数据集的汇总统计，那么这个批计算任务也就完成了。而流计算的数据是无边界的，数据会源源不断地产生。比如在线购物平台根据用户的浏览记录来进行实时推荐，如果用户浏览了A品牌的奶粉，平台可能会给他推荐B品牌的奶粉或是纸尿裤。用户的浏览记录就是源源不断产生的，而这个计算（给用户推荐什么）也会一直持续下去。

Why：为什么需要流计算？

传统的批计算针对的是有边界数据的处理，根据输入数据集的全集，计算出一个准确无误的结果。这种计算，是基于全量数据的计算，而流计算处理的是无边界的数据，我们不可能拿到全量数据。幸运的是，我们在处理无边界数据的时候，也是针对已经产生的数据进行计算，比如每隔一小时统计该时间段内的设备异常检测结果，我们只需对一小时的异常检测记录进行汇总。我们也不可能获取到未来的数据。批计算在处理这种场景的时候就可以通过将无边界的数据按固定时间间隔切分成一个个有边界数据进行计算。如下图所示。

这是使用批计算来处理无边界数据一个可能的做法。但是这个方法也有局限性，如果数据晚到了怎么办？原本属于[10:00-11:00)窗口的数据可能因为网络原因11:01才到，这时候[10:00-11:00)窗口的结果已经算好，这样必然造成计算结果的不正确。另一种场景，如果是对用户活动会话时间内的数据做处理，那就无法用固定时间间隔的方式进行切分，因为每一次切分都很可能会将某些会话一分为二，如下图所示。

当然还有时效性方面的考量，如果这个窗口大小是1分钟，甚至是1秒钟（每秒更新一次统计结果），传统的批处理是否能承受这种微批量方式，并满足低延时的要求？

How：流计算如何处理数据？

在了解流计算如何处理数据之前，我们先介绍几个相关的概念。

事件发生时间（Event Time）：事件真实发生的时间。实际情况中，许多流数据的处理会根据事件发生时间对数据进行处理。比如一笔信用卡交易记录，它的事件发生时间是这笔交易完成的时间。

处理时间（Process Time）：流计算引擎处理事件的时间。

在现实场景中，处理时间总是晚于事件发生时间。由于源系统、网络等多方面可能出现故障，理论上，事件有可能在它产生后的任意时间内到达处理引擎。因此，事件到达处理引擎时可能是乱序的；事件的发生时间和处理时间之间的偏差可能是任意大小。

接下来，我们通过Google提出的Dataflow模型来解答流计算引擎是如何处理数据的。Dataflow模型将无边界数据处理所要解决的问题抽象为4个问题——What、Where、When、How。这个模型虽然不是流计算数据处理的标准，但它奠定了流计算数据处理的理论基础，许多流计算引擎或多或少地采用或借鉴了它的思想和原理。

1. What results are calculated?

对无边界数据处理想获得的结果是什么，这个问题可以通过数据处理的计算逻辑来回答，可能是数据清洗、整合，可能是模式识别，也可能是指标计算等等。比如我们希望获得的是总交易额，那么我们会将每一笔交易金额进行累加。

2. Where in event time are results calculated?

在事件发生时间的哪个位置计算结果？换句话说，就是要计算哪些数据。比如我们想统计过去1小时的交易总额，那么这个where指的就是过去1小时到当前的这段时间内发生的交易。流计算引擎通过窗口化（windowing）的方式来解决这个问题。窗口化是通过划定一个边界将数据切分成一小块有限数据的方式进行处理。通过窗口化，我们就可以界定哪个位置范围内的数据是要用来做计算的。常用的窗口切分策略有以下3种：

• 固定窗口（Fixed windows，也称Tumbling windows）：将数据按照固定时间间隔进行切分，窗口和窗口之间不存在数据重叠。比如每隔一小时统计一次交易记录，那么我们可能会将数据切分为[10:00, 11:00), [11:00, 12:00), [12:00, 13:00)等多个窗口。
• 滑动窗口（Sliding windows，也称Hopping windows）：给定一个滑动窗口大小x和一个滑动周期y，将数据切分成多个窗口。当y<x时，窗口和窗口之间存在数据重叠；当y>x时，窗口和窗口之间会留下空挡，未被窗口覆盖的数据将不会被计算到；当y=x时，切分的结果正好和固定窗口是一致的。比如每隔5分钟统计一次过去1小时的交易总额，那么我们可能会将数据切分为[10:00, 11:00), [10:05, 11:05), [10:10, 11:10)等多个窗口。
• 会话窗口（Session windows）：通过一系列不活动的间隔来划分会话窗口。给定一个超时时间，任何发生时间间隔小于超时时间的2个事件属于同一个会话窗口。会话窗口可以用来分析用户行为，比如在一个网页点击流中，我们可以对用户在一个会话窗口中的点击行为进行分析，了解用户点击了多少商品，商品和商品之间的关联关系等等。
  

当然这个问题不仅指基于事件发生时间的数据处理，基于处理时间的数据处理也同样适用，对于后者，可以将事件的处理时间“看作”是发生时间。

3. When in processing time are results materialized？

在引擎处理时间上，何时计算结果并发给下游？知道要计算哪些数据后，就得知道什么时候计算。通过触发器（Triggers）来告诉引擎什么时候计算结果。

常用的触发器有两类。

第一类是反复更新的触发器。它随着窗口中数据的到来，周期性地产生一个更新后的窗格（Pane）。窗格是一个窗口内的一个特定输出，比如我们配置了3条数据到来就触发一次更新，假设窗口[10:00, 11:00)到达的前3条数据是[1,9,30],那么输出的这个[1,9,30]便是一个窗格。这一类触发器中有2种常用的触发器：计数触发器（Count Trigger），当接收到指定数量的数据时触发一次更新；处理时间触发器（Process Time Trigger），根据处理时间设定一个更新周期，比如每分钟触发一次更新。

第二类是完整性触发器。这类触发器仅当引擎被告知该窗口的输入已经完成时，产生一个窗格。由于数据在处理时间和事件发生时间之间的偏差，当我们希望处理的窗口能包含对应事件发生时间段内的完整数据时，可以采用这类完整性触发器。比如事件发生时间窗口[12:00,12:02)，当引擎被告知发生在[12:00,12:02)时间段内的数据都到达时，会触发窗口数据的输出。Flink中的EventTimeTrigger就属于完整性触发器。

那引擎又是如何被告知某个窗口的数据已经完整了呢？——答案是水位线（Watermarks）。水位线是引擎从事件发生时间的角度衡量数据在一个流中的处理进度和完整性的方式。我们来看下面这个图，横轴表示事件的发生时间，纵轴表示处理时间，理想情况下两者的关系应该是图中黑色的虚线，这意味着事件发生的同一时刻引擎便能处理数据。但现实情况是图中的红线，处理时间和事件时间总是存在一定的偏差，而且这个偏差是不确定的。这条红线就可以近似地看成是水位线。水位线可以看做是一个函数，自变量是处理时间（以及该时刻对应的数据、状态等），因变量是事件发生时间。对于一个给定的处理时间P1，水位线函数返回一个事件发生时间E1，引擎据此来判断，在P1时刻所有事件发生时间早于E1的数据已经全部到来。

水位线有两种，一种是完美的水位线，它能完美地反映数据的到达情况，不会有延迟数据（水位线认为某个窗口数据已经完整了，但之后仍有该窗口的数据到达）的情况发生。另一种是启发式的水位线，它根据一定的方法来猜测数据的到达情况，结果并不是完全准确的，可能会存在延迟数据。完美的水位线在现实情况中难以构造，且可能导致产生结果的延时增大；启发式水位线在实际应用中更易构造，且有更低的延时，但缺点是可能导致部分数据未被处理，产生的结果有误。可参考下图示例。

除了使用单一触发器，也可以根据需要组合不同的触发器来使用。通过组合反复更新的触发器和完整性触发器，可以处理如下三种常见的场景：

• 在窗口中包含的数据完整到来之前，通过反复更新触发器周期性地产生窗格以便快速得知窗口当前数据的计算结果；
• 在窗口完成时通过完整性触发器获取窗口计算结果；
• 在有延迟数据到来时，通过反复更新触发器周期性地产生窗格以更新窗口的计算结果。在使用启发式水位线的情况下，可能会存在延迟数据的情况，可以采用这种方式来处理延迟数据。多数情况下，对窗口数据和状态的缓存会消耗系统资源（内存、存储等），很难无限期等待延迟数据的到来，因此在实际使用中，一般还会对延迟数据设置一个最大延迟，超过这个最大延迟的数据将会被丢弃。
  
  

4. How do refinements of results relate?

如何优化计算结果？一个窗口的数据根据配置的触发器的不同，可能会产生多个窗格，每窗格产生时都会触发一次结果的计算，那当一个新的窗格产生时，我们如何根据前一次结果和新的窗格中的数据来计算窗口的新结果呢？根据实际应用的需要，可能的处理方式有三种：

丢弃（Discarding）：每次新的窗格到来时，上一次窗格的计算结果会被丢弃，仅使用当前窗格的数据来计算结果。当计算结果的下游希望能获取每次触发的窗格的独立结果时，可以使用这种方式。

累积（Accumulating）：将新窗格的数据累积到上一次窗格的计算结果中，得出新的计算结果。当计算结果的下游希望每次得到的是这个窗口当前所有到达数据计算的累积值时，使用这种方式。这是最常用的一种方式。
累加并回撤（Accumulating and Retracting）：在累积的基础上增加对上一次结果的回撤。当计算结果的下游将结果进行重新分组处理时，上一次的结果和更新的结果可能会被分发到不同的分组进行处理，可以采用这种方式将上一次的结果回撤，以保证数据准确性。

上面描述的4个问题中有很多是和时间相关的数据处理，我们在现实中还有很多流计算应用场景是与时间无关的数据处理，它们不涉及窗口、触发、结果优化等复杂处理。这些数据处理的过程不关心事件的发生时间或处理时间，不会存在延迟、乱序等问题，只需要每次处理一个事件即可。比如数据过滤，银行可能需要将金额大于5万的交易实时上报给监管机构，这个计算过程仅需根据交易金额大小判断。与时间无关的场景的数据处理也可以看做是上述4个问题的特例，比如窗口可以是大小为1的计数窗口，触发器是数量为1的计数触发器。

说了这么多，其实都是理论基础，虽然许多流计算引擎参考了Dataflow模型，但在实际实现上还存在许多差异，所以当你发现其它引擎的实现和我们这里的描述不一样时，请不要诧异。

后记

本文介绍的只是流计算如何处理数据，具体每个不同的流计算引擎到底处理得“好不好”，比如性能、一致性保障、状态管理等表现如何，还需要一一分解。后续笔者将选取几个使用较为广泛的流计算引擎进行剖析，敬请期待。