本文选自《实时流计算系统设计与实现》 文末有惊喜
状态管理是流计算系统的核心问题之一。在实现流数据的关联操作时，流计算系统需要先将窗口内的数据临时保存起来，然后在窗口结束时，再对窗口内的数据做关联计算。在实现时间维度聚合特征计算和关联图谱特征计算时，更是需要创建大量的寄存用于记录聚合的结果。而CEP的实现，本身就与常说的有限状态机（Finite-state machine，FSM）是密切相关的。不管是为了关联计算而临时保存的数据，还是为了保存聚合计算的数据，抑或是CEP里的有限状态机，这些数据都是流计算应用开始运行之后才创建和积累起来。如果没有做持久化操作，这些数据在流计算应用重启后会被完全清空。正因为如此，我们将这些数据称之为流计算应用的“状态”。从各种开源流计算框架的发展历史来看，大家对实时流计算中的“状态”问题也是一点点逐步弄清楚的。
我们将流在执行过程中涉及到的状态分为两类：流数据状态和流信息状态。
- 流数据状态。在流数据处理的过程中，可能需要处理事件窗口、时间乱序、多流关联等问题，在解决这些问题的过程中，通常会涉及到对部分流数据的临时缓存，并在处理完后将其清理。我们将临时保存的部分流数据称为“流数据状态”。
- 流信息状态。在对流数据的分析过程中，会得到一些我们感兴趣的信息，比如时间维度的聚合数据、关联图谱中的一度关联节点数、CEP中的有限状态机等，这些信息可能会在后续的流数据分析过程中被继续使用，从而需要将这些信息保存下来。同时在后续的流数据处理过程中，这些信息还会被不断地访问和更新。我们将这些分析所得并保存下来的数据称为“流信息状态”。
图1: 流数据状态和流信息状态
为什么区分这两种状态非常重要？思考这么一个问题，如果我们要计算“用户过去7天交易的总金额”，该如何做？一种显而易见的方法，是直接使用在各种流计算框架中都提供的窗口函数来实现。比如在Flink中如下：
userTransactions.keyBy(0)// 滑动窗口，每1秒钟计算一次7天窗口内的交易金额.timeWindow(Time.days(7), Time.seconds(1)).sum(1);上面的Flink示例代码使用timeWindow窗口，每1秒钟计算一次7天窗口内的总交易金额。其它流计算平台如Spark Streaming、Storm等也有类似的方法。但这样做有以下几点非常不妥：
- 这个计算是每1秒钟才能输出结果，而如果是需要每来一个事件就要计算一次该事件所代表的用户在“过去7天交易的总金额”，这种做法显然就不可行。
- 窗口为7天，滑动步长为1秒，这两个时间相差的数量级也太大了。这也意味着需要在“7天除以1秒”这么多个窗口中被重复计算！当然，这里设置1秒是因为要尽可能地“实时”。如果觉得1秒太“过分”，也可以设置滑动步长为30秒、60秒等，但这并不能改变重复计算的本质，且滑动步长越长，离“实时计算”越远。
- 窗口为7天，就需要在实时流计算系统中缓存7天的流数据。而我们想要得到的其实只是一个聚合值而已，所以保存7天完整的流数据似乎有些杀鸡用牛刀。当然，Flink对诸如sum、max、min之类的窗口聚合计算做了优化，可以不用保存窗口里的全部数据，只需要保留聚合结果即可。但是如果用户需要做些定制化操作（比如自定义Evictor）的话，就需要保存窗口内的全量数据了。
- 如果要在一个事件上，计算几十个类似于“用户过去7天交易的总金额”这样的特征，按照timeWindow的实现方法，每个特征可能会有不同的时间窗口和滑动步长，该怎样同步这几十个特征计算的结果呢？
所以说，直接使用由流计算框架提供的窗口函数来实现诸如“时间维度聚合特征”的计算问题，我们在很多情况下都会遇到问题。究其根本原因，是因为混淆了“对流的管理”和“对数据信息的管理”这两者本身。因为“窗口”实际上是对“流数据”的分块管理，我们用“窗口”来将“无穷无尽”的流数据分割成一个个的“数据块”，然后在“数据块”上做各种计算。这属于对流数据的“分而治之”处理。我们不能将这种针对“流数据”本身的分治管理模式，与我们对数据的业务信息分析窗口耦和起来。
因此，我们需要将“对流的管理”和“对数据信息的管理”这两者分离开来。其中“对流的管理”需要解决诸如窗口、乱序、多流关联等问题，其中也会涉及对数据的临时缓存，它缓存的是流数据本身，因此我们称之为“流数据状态”。而“对数据信息的管理”则是为了在我们在分析和挖掘数据内含信息时，帮助我们记录和保存业务分析结果，因而称之为“流信息状态”。
流数据状态管理中，比较重要的就是事件窗口、时间乱序和流的关联操作。
事件窗口是产生流数据状态的主要原因。比如“每30秒钟计算一次过去五分钟交易总额”、“每满100个事件计算平均交易金额”、“统计用户在一次活跃期间点击过的商品数量”等。对于这些以“窗口”为单元来处理事件的方式，我们需要用一个缓冲区（buffer）临时地存储过去一段时间接收到的事件，等触发窗口计算的条件满足时，再触发处理窗口内的事件。当处理完成后，还需要将过期和以后不再使用的数据清除掉。另外，在实际生产环境中，可能会出现故障恢复、重启等情况，这些“缓冲区”的数据在必要时需要被写入磁盘，并在重新计算或重启时恢复。
解决时间乱序问题是使用流数据状态的另一个重要原因。由于网络传输和并发处理的原因，在流计算系统接收到事件时，非常有可能事件已经在时间上乱序了。比如时间戳为1532329665005的事件，比时间戳为1532329665001的事件先到达流计算系统。怎样处理这种事件在时间上乱序的问题呢？通常的做法就是将收到的事件先保存起来，等过一段时间后乱序的事件到达时，再将其和保存的事件按时间排序，这样就恢复了事件的时间顺序。当然，上面的过程存在一个问题，就是“等过一段时间”到底是怎样等以及等多久？针对这个问题有一个非常优秀的解决方案，就是水印（watermark）。使用水印解决时间乱序的原理如下，在流计算数据中，按照一定的规律（比如以特定周期）插入“水印”，水印是一个时间戳，当处理单元接收到“水印”时，表示应该处理所有时间戳在该水印之前的事件。我们通常将水印设置为事件的时间戳减去一段时间的值，这样就给先到的时间戳较大的事件一个等待晚到的时间戳较小的事件的机会，而且确保了不会没完没了地等待下去。
流的关联操作也会涉及流数据状态的管理。常见的关联操作有join和union。特别是在实现join操作时，需要先将参与join操作的各个流的相应窗口内的数据缓存在流计算系统内，然后以这些窗口内的数据为基础，做类似于关系型数据库中表与表之间的join计算，得到join计算的结果，之后再将这些结果以流的方式输出。很显然，流的关联操作也是需要临时保存部分流数据的，故而也是一种“流数据状态”的运用。
除了以上三种“流数据状态”的主要用途外，还有些地方也会涉及流数据状态的管理，比如排序（sorting）、分组（group by）等。但不管怎样，这些操作都有个共同的特点，即它们需要缓存的是部分原始的流数据。换言之，这些操作要保存的状态是部分“流数据”本身。这也正是将这类状态取名为“流数据状态”的原因。流信息状态是为了记录流数据的处理和分析过程中获得的我们感兴趣的信息，这些信息会在后续的流处理过程中会被继续使用和更新。以“实时计算每个交易事件在发生时过去7天交易的总金额”这个计算为例，可以将每小时的交易金额记录为一条状态，这样，当一个交易事件到来时，计算“过去7天的交易总金额”，就是将过去7天每个小时的总交易金额读取出来，然后对这些金额记录求总和即可。在上面这个例子中，将每小时的交易金额记录为一条状态，就是我们说的“流信息状态”。
流信息状态的管理通常依赖于数据库完成。这是因为对于从流分析出来的信息，我们可能需要保存较长时间，而且数据量会很大，如果将这些信息状态放在内存中，势必会占用过多的内存，这是不必要的。对于保存的流信息状态，我们并不是在每次计算中都会用到，它会存在冷数据和过期淘汰的问题。所以，对于流信息状态的管理，交给专门的数据库是非常明智的。毕竟目前为止，各种数据库的选择十分丰富，而且许多数据库对热数据缓存和TTL机制都有非常好的支持。
实时流计算应用中的“流数据状态”和“流信息状态”。可以说是分别从两个不同的维度对“流”进行了管理。前者“流数据状态”是从“时间”角度对流进行管理，而后者“流信息状态”则是从“空间”角度对流的管理。“流信息状态”弥补了“流数据状态”弥补了“流数据状态”只是对事件在时间序列上做管理的不足，将流的状态扩展到了任意的空间。
作者简介：周爽，本硕毕业于华中科技大学，先后在华为2012实验室高斯部门和上海行邑信息科技有限公司工作。开发过实时分析型内存数据库RTANA、华为公有云RDS服务、移动反欺诈MoFA等产品。目前但任公司技术部架构师一职。著有《实时流计算系统设计与实现》一书。
本次联合【机械工业出版社华章公司】为大家送上1本作者的正版书籍《实时流计算系统设计与实现》
请在关注“实时流式计算” 并在后台回复 “抽奖”参与活动
更多实时数据分析相关博文与科技资讯，欢迎关注 “实时流式计算”
