流式统计的几个难点

流式统计听着挺容易的一个事情，说到底不就是数数嘛，每个告警系统里基本上都有一个简单的流式统计模块。但是当时基于storm做的时候，这几个问题还是困扰了我很长时间的。没有用过spark streaming/flink，不知道下面这些问题在spark streaming/flink里是不是都已经解决得很好了。

时间窗口切分问题

做流式统计首要的问题是把一个时间窗口内的数据统计到一起。问题是，什么是时间窗口？有两种选择

• 日志时间（event timestamp）
• 墙上时间（wall clock）

最简单的时间窗口统计的是基于“墙上时间”的，每过1分钟就切分出一个新窗口出来。比如statsd，它的窗口切分就是这样的。这种基于“墙上时间”的统计有一个非常严重的问题是不能回放数据流。当数据流是实时产生的时候，“墙上时间”的一分钟也就只会有一分钟的event被产生出来。但是如果统计的数据流是基于历史event的，那么一分钟可以产生消费的event数量只受限于数据处理速度。另外event在分布式采集的时候也遇到有快有慢的问题，一分钟内产生的event未必可以在一分钟内精确到达统计端，这样就会因为采集的延迟波动影响统计数据的准确性。实际上基于“墙上时间”统计需要

collection latency = wall clock - event timestamp

基于“墙上时间”的统计需要采集延迟非常小，波动也很小才可以工作良好。大部分时候更现实的选择是需要基于“日志时间”来进行窗口统计的。

使用“日志时间”就会引入数据乱序的问题，对于一个实时event stream流，其每个event的timestamp未必是严格递增的。这种乱序有两种因素引入：

• event产生的机器的时钟不完全同步（NTP有100ms左右的不同步）
• event从采集到到达kafka的速度不均衡（不同的网络线路有快有慢）

我们希望的流式统计是这样的：

但是实际上数据只是基本有序的，也就是在时间窗口的边缘会有一些event需要跨到另外一个窗口去：

最简单的分发event到时间窗口代码是这样的

window index = event timestamp / window size

对1分钟的时间窗口 window size 就是60，timestamp除以60为相同window index的event就是在同一个时间窗口的。问题的关键是，什么时候我可以确信这个时间窗口内的event都已经到齐了。如果到齐了，就可以开始统计出这个时间窗口内的指标了。然后突然又有一个落后于大伙的event落到这个已经被计算过的时间窗口如何处理？

• 对于大部分统计而言，一个时间窗口统计出多条结果存入db并不是什么大的问题，从db里查询的时候把多条结果再合并就可以了。
• 对于一些类型的统计（非monad），比如平均值，时间窗口内的event分为两批统计出来的结果是没有办法被再次汇总的。
• 实时类的计算对时间敏感，来晚了的数据就没有意义了。比如告警，一个时间窗过去了就没有必要再理会这个时间窗口了。

所以对于来晚了的数据就两种策略：要么再统计一条结果出来，要么直接丢弃。要确定什么时候一个时间窗口内的event已经到齐了，有几种策略：

• sleep 等待一段时间（墙上时间）
• event timestamp超过了时间窗口一点点不关闭当前时间窗口，而是要等event timestamp大幅超出时间窗口的时候才关闭窗口。比如12:05:30秒的event到了才关闭12:04:00 ~ 12:05:00的时间窗口。
• 一两个event超出了时间窗口不关闭，只有当“大量”的event超出时间窗口才关闭。比如1个event超过12:05分不关闭，如果有100个event超过了12:05的时间窗口就关闭它。

三种策略其实都是“等”，只是等的依据不同。实践中，第二种策略也就是根据“日志时间”的等待是最容易实现的。如果对于过期的event不是丢弃，而是要再次统计一条结果出来，那么过期的窗口要重新打开，又要经过一轮“等待”去判断这个过去的窗口什么时候再被关闭。

在spark上已经有人做类似的尝试了： Building Big Data Operational Intelligence platform with Apache Spark - Eric Carr (Guavus)

多流合并的问题

一个kafka的partition就是一个流，一个kafka topic的多个partition就是多个独立的流（offset彼此独立增长）。多个kafka topic显然是多个独立的流。流式统计经常需要把多个流合并统计到一起。这种里会遇到两个难题

• 多个流的速度不一样，如何判断一个时间窗口内的event都到齐了。如果按照前面的等待策略，可能处理一个流内部的基本有序局部乱序是有效的，但是对于多个流速差异很大的流就无能为力了。一个很快的流很容易把时间窗口往后推得很远，把其他流远远跑到后面。
• 流速不均不能靠下游兜着，下游的内存是有限的。根本上是需要一种“背压”的机制，让下游通知流速过快的上游，你慢点产生新的event，等等其他人。

举一个具体的例子：

spout 1 emit 12:05 spout 1 emit 12:06 spout 2 emit 12:04 spout 1 emit 12:07 spout 2 emit 12:05 // this is when 12:05 is ready 

要想知道12:05这个时间窗的event都到齐了，首先要知道相关的流有几个（在这例子里是spout1和spout2两个流），然后要知道什么时候spout1产生了12:05的数据，什么时候spout2产生了12:05的数据，最后才可以判断出来12:05的数据是到齐了的。在某个地方要存一份这样的流速的数据去跟踪，在窗口内数据到齐之后发出信号让相关的下游往前推动时间窗口。考虑到一个分布式的系统，这个跟踪要放在哪个地方做，怎么去通知所有的相关方。

极端一些的例子

spout 1 emit 13:05 spout 2 emit 12:31 spout 1 emit 13:06 spout 2 emit 12:32 

多个流的流速可能会相差到半个小时以上。考虑到如果用历史的数据汇入到实时统计系统里时，很容易因为计算速度不同导致不同节点之间的处理进度不一致。要计算出正确的结果，下游需要缓存这些差异的半个小时内的所有数据，这样很容易爆内存。但是上游如何感知到下游要处理不过来了呢？多个上游之间又如何感知彼此之间的速度差异呢？又有谁来仲裁谁应该流慢一些呢？

一个相对简单的做法是在整个流式统计的分布式系统里引入一个coordinator的角色。它负责跟踪不同流的流速，在时间窗口的数据到齐之后通知下游flush，在一些上游流速过快的时候（比如最快的流相比最慢的流差距大于10分钟）由coordinator发送backoff指令给流速过快的上游，然后接到指令之后sleep一段时间。一段基本堪用的跟踪不同流流速的代码： https://gist.github.com/taowen/2d0b3bcc0a4bfaecd404

数据一致性问题

低档一些的说法是这样的。假设统计出来的曲线是这样的：

如果中间，比如08:35左右重启了统计程序，那么曲线能否还是连续的？

高档一些的说法是，可以把流式统计理解为主数据库与分析数据库之间通过kafka消息队列进行异步同步。主数据库与分析数据库之间应该保持eventual consistency。

要保证数据不重不丢，就要做到生产到kafka的时候，在主数据库和kafka消息队列之间保持一个事务一致性。举一个简单的例子：

用户下了一个订单 主数据库里插入了一条订单的数据记录 kafka消息队列里多了一条OrderPlaced的event 

这个流程中一个问题就是，主数据插入成功了之后，可能往kafka消息队列里enqueue event失败。如果把这个操作反过来

用户下了一个订单 kafka消息队列里多了一条OrderPlaced的event 主数据库里插入了一条订单的数据记录 

又可能出现kafka消息队列里enqueue了，但是主数据库插入失败的情况。就kafka队列的目前的设计而言，对这个问题是无解的。一旦enqueue的event，除非过期是无法删除的。

在消费端，当我们从kafka里取出数据之后，去更新分析数据库的过程也要保持一个分布式事务的一致性。

取出下一条OrderPlaced evnet（指向的offset+1） 当前时间窗的统计值+1 重复以上过程，直到窗口被关闭，数据写入到分析数据库 

kafka的数据是可以重放的，只要指定offset就可以把这个offset以及之后的数据读取出来。所谓消费的过程就是把客户端保存的offset值加1的过程。问题是，这个offset指针保存在哪里的问题。常规的做法是把消费的offset保存到zookeeper里。那么这就有一个分布式的一致性问题了，zookeeper里offset+1了，但是分析数据库并没有实际把值统计进去。考虑到统计一般不是每条输入的event都会更新分析数据库，而是把中间状态缓存在内存中的。那么就有可能消费了成千上万个event，状态都在内存里，然后“啪”的一下机器掉电了。如果每次读取event都移动offset的话，这些event就丢掉了。如果不是每次都移动offset的话，又可能在重启的时候导致重复统计。

搞统计的人在乎这么一两条数据吗？其实大部分人是不在乎的。不少团队压根连offset都不保存，每次开始统计直接seek到队列的尾部开始。实时计算嘛，实时最重要了。准确计算？重放历史？这个让hadoop搞定就好了。但是如果就是要较这个真呢？或者我们不追求严格的强一致，只要求重启之后曲线不断开那么难看就好了。

别的流式计算框架不清楚，storm的ack机制是毫无帮助的。

storm的ack机制是基于每个message来做的。这就要求如果做一个每分钟100万个event的统计，一分钟就要跟踪100万个message id。就算是100万个int，也是一笔相当可观的内存开销。要知道，从kafka里读出来的event都是顺序offset的，处理也是顺序，只要记录一个offset就可以跟踪整个流的消费进度了。1个int，相比100万个int，storm的per message ack的机制对于流式处理的进度跟踪来说，没有利用消息处理的有序性（storm根本上假设message之间是彼此独立处理的），而变得效率低下。

要做到强一致是很困难的，它需要把

• 更新保存的offset
• 更新插入分析数据库

变成一个原子事务来完成。大部分分析数据库都没有原子性事务的能力，连插入三条数据都不能保持同时变为可见，且不说还要用它来记录offset了。考虑到kafka在生产端都无法提供分布式事务，event从生产出来就不是完全一致的（多产生了或者少产生了），真正高一致的计费场景还是用其他的技术栈。所以值得解决的问题是，如何在重启之后，把之前重启的时候丢弃掉的内存状态重新恢复出来，使得统计出来的曲线仍然是连续的。

解决思路有三点：

• 上游备份策略：重启的时候重放kafka的历史数据，恢复内存状态
• 中间状态持久化：把统计的状态放到外部的持久的数据库里，不放内存里
• 同时跑两份：同时有两个完全一样的统计任务，重启一个，另外一个还能正常运行。

内存状态管理的问题

做流式统计的有两种做法：

• 依赖于外部存储管理状态：比如没收到一个event，就往redis里发incr增1
• 纯内存统计：在内存里设置一个counter，每收到一个event就+1

基于外部存储会把整个压力全部压到数据库上。一般来说流式统计的流速是很快的，远大于普通的关系型数据库，甚至可能会超过单台redis的承载。这就使得基于纯内存的统计非常有吸引力。大部分的时候都是在更新时间窗口内的内存状态，只有当时间窗口关闭的时候才把数据刷到分析数据库里去。刷数据出去的同时记录一下当前流消费到的位置（offset）。

这种纯内存的状态相对来说容易管理一些。计算直接是基于这个内存状态做的。如果重启丢失了，重放一段历史数据就可以重建出来。

但是内存的问题是它总是不够用的。当统计的维度组合特别多的时候，比如其中某个字段是用户的id，那么很快这个内存状态就会超过单机的内存上限。这种情况有两种办法：

• 利用partition把输入的input分割，一个流分成多个流，每个统计程序需要跟踪的维度组合就变少了
• 把存储移到外边去

简单地在流式统计程序里开关数据库连接是可以解决这个容量问题的：

但是这种对外部数据库使用不小心就会导致两个问题：

• 处理速度慢。不用一些批量的操作，数据库操作很快就会变成瓶颈
• 数据库的状态不一直。内存的状态重启了就丢失了，外部的状态重启之后不丢失。重放数据流就可能导致数据的重复统计

但是这种把窗口统计的中间状态落地的好处也是显而易见的。重启之后不用通过重算来恢复内存状态。如果一个时间窗口有24小时，重算24小时的历史数据可能是很昂贵的操作。

版本跟踪，批量等都不应该是具体的统计逻辑的实现者的责任。理论上框架应该负责把冷热数据分离，自动把冷数据下沉到外部的存储，以把本地内存空闲出来。同时每次小批量处理event的时候都要记录处理的offset，而不是要等到窗口关闭等待时候。

数据库状态和内存状态要变成一个紧密结合的整体。可以把两者的关系想象成操作系统的filesystem page cache。用mmap把状态映射到内存里，由框架负责什么时候把内存里的变更持久化到外部存储里。

总结

基于storm做流式统计缺乏对以下四个基本问题的成熟解决方案。其trident框架可能可以提供一些答案，但是实践中好像使用的人并不多，资料也太少了。可以比较自信的说，不仅仅是storm，对于大多数流式计算平台都是如此。

• 时间窗口切分的问题
• 多流合并的问题
• 数据一致性问题（重启之后曲线断开的问题）
• 内存状态管理问题

这些问题要好好解决，还是需要一番功夫的。新一代的流式计算框架比如spark streaming/flink应该有很多改进。即便底层框架提供了支持，从这四个角度去考察一下它们是如何支持的也是非常有裨益的事情。