从S4到Storm（一）：当分布式遇上实时计算

你好我是程序员贵哥。到Spanner为止我们已经把大数据里关于数据存储和在线服务的重要论文解读完了。从这一讲开始我们就要开始讲解另一个重要的主题也就是大数据的流式处理。今天我们解读的第一篇论文来自一个曾经辉煌但是今天已经逐渐销声匿迹的公司Yahoo。这篇论文就是《S4:Distributed Stream Computing Platform》伴随着这篇论文的同样是一个开源系统Apache S4。和同样孵化自Yahoo的Hadoop不同S4虽然是最早发布的开源分布式流式数据处理系统但是在市场上最终却没有占有一席之地。不过学习S4的论文本身还是很有价值的。一方面你可以看到大数据流式处理的基础结构是怎么样的你会发现它和批处理的MapReduce是非常相似的另一方面它又会面临比批处理的MapReduce多得多的挑战和困难。那在学完这一讲之后你能有这样几点收获首先是对于大数据的流式计算问题的基本抽象也就是它在逻辑上应该是一个什么样的模型它想要解决的具体问题是什么。其次是S4这个系统的设计是怎么样的它是如何进行分布式计算的。最后是S4系统的缺陷有哪些以及有哪些问题S4干脆是回避了。这一点对于我们后面去认识Storm、Kafka这些系统的核心价值点非常关键。好了那接下来就请和我一起进入流式计算的世界吧。实时计算到底有多“实时”在MapReduce出现之后整个大数据处理领域就红火了起来Hadoop这样的开源项目也很快深入到各大互联网公司。一方面MapReduce帮我们解决了海量数据处理的问题但是另一方面我们常常又觉得MapReduce不太够用因为很多时候我们希望能够更加“实时”地处理数据。之所以我们希望能够“实时”进行数据处理是由于“效益”的原因。MapReduce一开始的主要的应用领域就是广告和搜索。通过MapReduce程序我们可以分析海量的用户搜索行为、广告点击行为来帮助我们优化搜索排序和广告展示。比如在广告领域最简单的一个优化的办法就是统计新广告的点击率点击率太低的广告我们不予展示把展示机会给别的广告就好了。在论文里Yahoo也举了这样一个例子通过在线统计点击率把低质量的广告过滤掉S4可以帮助他们再提升3%的点击率并且对广告收入没有任何影响。这个需求和我们通过MapReduce生成各种报表是一回事儿。唯一的一个差异在于**我们希望这个数据统计的反馈的时间能够尽量短一些也就是“实时一点”。**而这个差异就触到了MapReduce的痛点了。一般来说我们的MapReduce都是定时执行的比如每天运行一次生成一个报表或者频繁一点每小时运行一次计算上一个小时的点击率数据。但是这个获得反馈数据的频率还是太慢了。每小时运行一次MapReduce程序意味着我们的统计数据平均要晚上半个小时。而半个小时里低质量的广告或者搜索结果已经曝光了很多次了。举个例子现在无论是什么样的社会热点新闻很容易在微博热搜上出现。往往在新闻发生后的一两分钟就已经有大量的搜索出现在微博热搜里了。如果我们要等待半个小时才能统计到这些搜索那么热搜功能就可以说形同虚设了。所以我们希望能在尽可能短的时间内就得到这样的反馈数据。那么你可能要问了我们能不能直接更频繁地运行MapReduce程序呢比如每分钟运行一次是不是就解决这个问题了如果你仔细学习了之前的课程相信你自己也会意识到这样是行不通的。采用频繁运行MapReduce程序的办法我们至少会遇到两个问题。一个是大量的“额外开销”。我们之前讲过MapReduce的额外开销不小再小的任务也需要个十几秒到一分钟的运行时间。如果我们高频率每分钟运行MapReduce任务那么“额外开销”占的时间比重和硬件资源会非常高也很浪费。二个是我们不得不让输入文件变得极其“碎片化”。无论是GFS还是HDFS都是把文件变成一个个64MB大小的Block然后MapReduce通过分布式并行读取来进行快速分析。但是如果我们需要每分钟都处理数据那么对于输入的数据就要按照分钟进行分割。每分钟我们都需要有很多个文件分布到GFS/HDFS上不同的数据节点。这样我们的文件都会变得很小也就丧失了顺序读取大文件的性能优势。其实高频率地执行MapReduce还会有很多问题。而归根到底是这两点第一MapReduce是为“高吞吐量”而设计的一个系统。在整个系统设计理念里它没有考虑“低延时”这个需求。第二MapReduce的数据是一份“边界明确bounded”的数据。在进行数据处理之前要处理的数据已经存放在存储系统上了。而我们想要进行的实时数据统计想要处理的是一份“无边界unbounded”的数据会不断地有新数据流入进来永无停歇。所以我们需要一个全新的流式数据处理系统这也是S4这个系统的出发点。流式计算的逻辑模型我们先来看一看S4是怎么抽象我们的流式计算的。S4把所有的计算过程都变成了一个个处理元素Processing Element对象简称为PE对象。我这里特地加上了对象就是因为在实现上PE就是一个面向对象编程里面一个实际的对象。每一个PE对象都有四部分要素组成分别是PE本身的功能functionality这个体现为PE类里实现的业务逻辑函数以及为了这个类配置的各种参数PE能够处理的事件类型types of eventsPE能够处理的事件的键keyed attributePE处理的事件的键对应的值value。对于流式的数据处理就是由一个个PE组成的有向无环图DAG。有向无环图的起点是一些特殊的被称为无键PEKeyless PE的对象。这些对象的作用其实就是接收外部发送来的事件流这些外部发送过来的事件流其实就是一条条的消息。这些无键PE会解析对应的消息变成一个个事件。然后给每个事件打上三个信息分别是事件类型Event Type事件的Key事件的Value。然后可以把事件给发送出去。接着下游的其他PE对象会根据自己定义的事件类型和能处理的键来接收对应的消息并且处理这个消息。如果当前系统里没有对应的键的PE那么系统会创建一个新的PE对象。处理数据的PE对象可以选择处理完之后立刻发送一个新的事件出去也可以选择在对象内部来维护一个状态然后当处理了一定数量的消息之后或者过了一个固定的事件间隔之后把消息发送出去。最后在整个有向无环图的终点会有一系列的PE对象。这些对象会把最终的计算结果**发布Publish。**这个发布的频率也和其他PE发送消息的逻辑类似可以在每收到一个事件就发送也可以要求接收到一定数量的事件或者每隔一个特定的时间间隔发送。这么来描述可能整个过程有点过于抽象我们还是来一起看一看论文里图一的示例。这个例子是用来统计整个系统里出现得最多的K个单词也就是Top K它的整个DAG的结构是这样的首先在DAG的起始节点是一个QuoteSplitterPE这个PE也是一个无键PE。它负责接收外部发送来的句子然后分割成一个个单词接着会统计单词在句子里面出现的次数。然后这个PE会把每个单词的出现次数作为一个WordEvent发送出去。对应的Event的Key就是Word具体单词的这么一个组合Tuple。而对应的Event的值就是Count出现次数的这么一个组合Tuple。第二层里是一系列叫做WordCountPE的PE对象。它在系统里面申明我只接收WordEvent。然后每个不同的单词都会有一个对应的PE对象。所以可以想象整个系统中会有海量的PE对象。它的逻辑也很简单上游的PE会把相同单词的WordEvent都发送到同一个PE那么这个PE里就可以统计到这个单词出现的总的次数。每当收到一个事件这个单词的出现次数就会更新对应的它就会向下游发送一个UpdatedCountEvent也就是更新单词计数的事件。这个事件里对应的Key是SortN这样一个组合每一个PE对象里的N都是随机的但是固定不变的。这个组合是为了下一层的负载均衡我们可以自己去设定N这个参数N越大意味着下游的PE对象越多负载就会分配到更多不同的对象里去计算的。而对应的值则包括了对应的单词是什么以及对应的单词的出现次数。也就是((Word具体单词), (Count出现次数)) 这么一个组合。第三层里则是一系列叫做SortPE的对象。它的作用则是接收上游不同单词的出现次数然后在内部进行排序。最后输出自己内部排序的Top K再给到下游。本质上它相当于是所有单词的某一个分区的数据。这个分区包含了一部分单词的所有数据。我们前面设定了N是几我们就会有几个SortPE的对象。给到下游的事件叫做PartialTopkEvent看名字你就知道它包含的信息就是一个部分数据的Top K。所有SortPE的对象输出的消息的Key都是相同的因为为了获得全局的排序它们需要发送给同一个PE对象。在这里这个Key就被写死成了topk1234 这么一个组合。而Value则是K个单词出现次数的集合。而整个DAG的终点则是唯一的一个MergePE。它的作用就是接收PartialTopkEvent然后在内部进行一次归并选出全局的TopK。并且最终它还需要把对应的数据写入到外部其他的存储系统比如数据库里供其他的应用读取。S4这个把整个数据处理流程变成一个有向无环图的设计也是后续所有流式处理系统都采用的一个解决方案。所有的数据变成了事件流而开发人员只需要做两件事情第一是设计整个DAG应该是什么样子的。第二是实现这个当中每一个节点的业务逻辑代码。而开发人员不需要关心数据是在哪里被处理的。这些都由S4这个分布式系统自己来决定。师从MapReduce的设计理念其实S4的系统架构和我们之前看过的MapReduce这样的框架一脉相承。PE其实和Map/Reduce函数一样只是一个抽象的概念。不过S4的系统设计要更加激进一点那就是S4选择了一个无中心的完全对称的架构。S4和我们之前看过的所有系统都不一样没有所谓的Master节点。如果一定要说有一个中心化的地方的话S4依赖于Zookeeper也就是一个类似于Chubby这样的分布式锁系统。S4的所有服务器都会作为一个处理节点ProcessingNode简称PN注册在Zookeeper上。具体如何分配负载是由各个节点协商决定的而不是由一个中心化的Master统一分配。每一个处理节点都是相同的它由上下两部分组成。上面是实际的业务处理逻辑模块它通过Event Listener监听外部发送过来的消息转发给对应的PE对象。PE对象的所有输出结果都发送给Dispatcher让Dispatcher确定应该发送给哪些PE里。实际的消息发送会由Dispatcher交给Emitter对外发送出去。业务处理模块里只会确定对应的消息发送应该发送给哪一个逻辑上的PE实际具体发送到哪一台物理节点则是由下面的通信层模块来决定的。这个模块主要解决这样几个问题首先是具体的路由也就是Event要去的某一个逻辑PE到底在哪台物理服务器上是由通信层模块来找到并且发送出去的上层的业务处理流程不需要知道。其次是负载均衡不同的单词更新的频率可能不一样。所以不同的处理节点的负载也会不一样。当有一个新的单词出现的时候我们需要判断新的PE应该放到哪一个节点上去。然后就是底层的容错恢复机制了当有特定节点挂掉的时候我们需要在其他的节点上恢复原先这个节点被分配的PE。最后就是实际的传输协议S4是一个“插件式”的架构也就是底层的传输协议也是可以切换的。S4既支持通过TCP发送消息确保消息能够发送成功也支持通过UDP发送消息来支持更大的吞吐量。你可以看到这个其实和我们看MapReduce的框架是类似的开发人员的关注点只需要在PE这个纯粹的业务逻辑层面。至于计算在哪一台服务器上发生各个节点之间是怎么通信的开发人员完全不需要关心。稍显“过时”的伸缩和容错能力不过看到这里相信你也会有一些疑问。以单词计数为例看起来一个S4在线上的有向无环图就需要有海量的对象这个数量级可能是数万乃至数十万。而不像之前我们看过的MapReduce那样只需要有少数Map和Reduce就好了。没错S4的设计其实还有粗糙也还有着很多的问题。首先就是这里的海量对象的问题。由于每一个处理数据的Key都要是一个对象系统里就会有海量的对象。而一个Key如果只出现一次之后再也不出现了也要占用内存。S4对此的解决办法是给Key设定TTL定期清理掉不需要的Key。其次是S4里没有时间窗口的概念。在我们进行实时数据处理的时候我们需要统计的常常是“过去一分钟的热搜”或者“过去一小时的热搜”这样有一个时间范围的数据。但是在S4的设计里我们并没有地方可以设定这个时间窗口。所以类似的需求需要我们自己在PE的代码里面去维护或者实现一下子大大增加了开发的难度和复杂度。第三是S4的容错处理非常简单。S4能够做到的容错其实就是某一个计算节点挂掉了我们重新再起一个计算节点承担它的工作。但是原先节点里所有PE维护的状态信息就都丢失了。我们既不知道目前的统计信息是什么也不知道目前处理到哪些事件了。Yahoo给出的答案是退回到离线批处理计算的数据上但是这个显然就不满足流式处理一开始的需求了。只能算是个聊胜于无的方案。最后一个问题则是S4虽然是一个分布式系统但是并不支持真正的动态扩容。在一开始论文的假设部分就假设了运行中的集群不会增加或者减少节点。这样带来的问题就是当负载快速上升的时候S4的策略是随机丢弃一些数据本质上是对数据进行了采样而不是能够通过简单增加硬件来解决问题。不过不管怎么说S4还是让大数据的的流式处理迈出了第一步。而这些S4并没有回答好的问题也会为接下来的流式数据处理系统的兴旺拉开了帷幕。小结好了对于S4的论文我们到这里也就解读完了。我们看到随着大数据的价值深入人心MapReduce这样定期定时进行数据处理的方式逐渐难以满足业务需求。于是大数据的流式计算登上了历史舞台。Yahoo通过S4系统进行低延时的“实时”数据处理。整个系统的设计理念类似于MapReduce开发人员只需要实现Processing Element这样的业务处理逻辑而不需要关心“分布式”是怎么运行的。整个框架会完成数据的分发、计算节点的调度以及容错之后的恢复。通过S4Yahoo能够及时地获取广告和搜索数据的反馈以及进行在线的A/B测试。整个S4内部的设计也将业务逻辑层和网络协议、数据路由、负载均衡等拆分开来做成了一个可插拔Pluggable的系统架构。在整个的流式数据处理框架里S4采用了一个典型的Actor模式。整个数据处理的流程可以被画成一个有向无环图图里的每一个点都是一个处理元素每一条边都是一条消息传递的路径而每一个处理元素都会被托管在某一个处理节点里。处理元素负责实现业务逻辑并且可以保存计算结果在内存。同时S4支持你定时地将对应的结果发布到外部的存储系统里使得计算结果对外可用。但是S4的设计显然也是很粗糙的。S4采用了一个完全对称、没有中心节点的分布式架构虽然看起来这个解决了“单点故障”问题但是也因此放弃了动态扩容而只能在大量流量进入的时候选择服务降级的解决方案。而在业务层面S4的容错也只是考虑“计算节点”层面的容错。容错只是将挂掉的节点能够在其他的硬件上重新运行起来但是已经处理的历史数据都已经丢失了。而对于节点之间的数据传输S4也没有作出全链路的传输保障。这些问题也是后面的Storm、Kafka、Flink这些系统出现的出发点。而S4自己却在2014年就从Apache孵化的项目中“退役retried”了。