从Storm和Spark 学习流式实时分布式计算的设计

消息传递和分发

对于实现的逻辑来说，它们都是有向无环图的一个节点，那么如何设计它们之间的消息传递呢？或者说数据如何流动的？因为对于分布式系统来说，我们不能假定整个运算都是在同一个节点上（事实上，对于闭源软件来说，这是可以的，比如就是满足一个特定运算下的计算，计算平台也不需要做的那么通用，那么对于一个运算逻辑让他在一个节点完成也是可以了，毕竟节省了调度和网络传输的开销）。或者说，对于一个通用的计算平台来说，我们不能假定任何事情。

消息传递和分发是取决于系统的具体实现的。通过对比Storm和Spark，你就明白我为什么这么说了。

Spark的消息传递

对于Spark来说，数据流是在通过将用户定义的一系列的RDD转化成DAG图，然后DAG Scheduler把这个DAG转化成一个TaskSet，而这个TaskSet就可以向集群申请计算资源，集群把这个TaskSet部署到Worker中去运算了。当然了，对于开发者来说，他的任务是定义一些RDD，在RDD上做相应的转化动作，最后系统会将这一系列的RDD投放到Spark的集群中去运行。

 

Storm的消息传递      

对于Storm来说，他的消息分发机制是在定义Topology的时候就显式定义好的。也就是说，应用程序的开发者需要清楚的定义各个Bolts之间的关系，下游的Bolt是以什么样的方式获取上游的Bolt发出的Tuple。Storm有六种消息分发模式：

 

1. Shuffle Grouping: 随机分组，Storm会尽量把数据平均分发到下游Bolt中。
2. Fields Grouping：按字段分组， 比如按userid来分组， 具有同样userid的tuple会被分到相同的Bolt。这个对于类似于WordCount这种应用非常有帮助。
3. All Grouping： 广播， 对于每一个Tuple， 所有的Bolts都会收到。这种分发模式要慎用，会造成资源的极大浪费。
4. Global Grouping: 全局分组， 这个Tuple被分配到storm中的一个bolt的其中一个task。这个对于实现事务性的Topology非常有用。
5. Non Grouping: 不分组， 这个分组的意思是说stream不关心到底谁会收到它的tuple。目前这种分组和Shuffle grouping是一样的效果， 有一点不同的是storm会把这个bolt放到这个bolt的订阅者同一个线程里面去执行。
6. Direct Grouping: 直接分组,  这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。

 

消息传递要点

消息队列现在是模块之间通信的非常通用的解决方案了。消息队列使得进程间的通信可以跨越物理机，这对于分布式系统尤为重要，毕竟我们不能假定进程究竟是部署在同一台物理机上还是部署到不同的物理机上。RabbitMQ是应用比较广泛的MQ，关于RabbitMQ可以看我的一个专栏：RabbitMQ

提到MQ，不得不提的是ZeroMQ。ZeroMQ封装了Socket，引用官方的说法： “ZMQ (以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层，像框架一样的一个 socket library，他使得 Socket 编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ 的明确目标是“成为标准网络协议栈的一部分，之后进入 Linux 内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD 套接字之上的一层封装。ZMQ 让编写高性能网络应用程序极为简单和有趣。”

因此, ZeroMQ不是传统意义上的MQ。它比较适用于节点之间和节点与Master之间的通信。Storm在0.8之前的Worker之间的通信就是通过ZeroMQ。但是为什么0.9就是用Netty替代了ZeroMQ呢？说替代不大合适，只是0.9的默认的Worker之间的通信是使用了Netty，ZeroMQ还是支持的。Storm官方认为ZeroMQ有以下缺点：

 

1. 不容易部署，尤其是在云环境下：以为ZMQ是以Ｃ写的，因此它还是紧依赖于操作系统环境的。
2. 无法限制其内存。通过JVM可以很容易的限制java所占用的内存。但是ZMQ对于Storm来说是个黑盒似得存在。
3. Storm无法从ZMQ获取信息。比如Storm无法知道当前buffer中有多少数据为发送。

 

当然了还有所谓的性能问题，具体可以访问Netty作者的blog。结论就是Netty的性能比ZMQ（在默认配置下）好两倍。不知道所谓的ZMQ的默认配置是什么。反正我对这个结果挺惊讶。当然了，Netty使用Java实现的确方便了在Worker之间的通信加上授权和认证机制。这个使用ZMQ的确是不太好做。

高可用性

HA是分布式系统的必要属性。如果没有HA，其实系统是不可用的。那么如果实现HA？对于Storm来说，它认为Master节点Nimbus是无状态的，无状态意味着可以快速恢复，因此Nimbus并没有实现HA（不知道以后的Nimbus是否会实现HA，实际上使用ZooKeeper实现节点的HA是开源领域的通用做法）。为什么说Nimbus是无状态的呢？因为集群所有的元数据都保存到了ZooKeeper（ZK）中。Nimbus定时从ZK获取信息，并且通过向ZK写信息来控制Worker。Worker也是通过从ZK中获取信息，通过这种方式，Worker执行从Nimbus传递过来的命令。

Storm的这种使用ZK的方式还是很值得借鉴的。

Spark是如何实现HA的？我的另外一篇文章分析过Spark的Master是怎么实现HA的：Spark技术内幕：Master基于ZooKeeper的High Availability（HA）源码实现 。

也是通过ZK的leader 选举实现的。Spark使用了百行代码的级别实现了Master的HA，由此可见ZK的功力。

除了这些Master的HA，还有每个Worker的HA。或者说Worker的HA说法不太准确，因此对于集群里的工作节点来说，它可以非常容易失败的。这里的HA可以说是如何让Worker失败后快速重启，重新提供服务。实现方式也可以由很多种。一个简单的方法就是使用一个容器（Container）启动Worker并且监控Worker的状态，如果Worker异常退出，那么就重新启动它。这个方法很简单也很有效。

如果是节点宕机呢？上述方法肯定是不能用的。这种情况下Master会检测到Worker的心跳超时，那么就会从资源池中把这个节点删除。回到正题，宕机后的节点重启涉及到了运维方面的知识。对于一个集群来说，硬件宕机这种情况应该需要统一的管理，也就是集群也可以由一个Master，维持每个节点的心跳来确定硬件的状态。如果节点宕机，那么集群首先是重启它。如果启动失败可能会通过电话或者短信或者邮件通知运维人员。因此运维人员为了保证集群的高可用性付出了很多的努力，尤其是大型互联网公司的运维人员，非常值得点赞。当然了这个已经不是Storm或者Spark所能涵盖的了。

存储模型与数据不丢失

其实，数据不丢失有时候和处理速度是矛盾的。为了数据不丢失就要进行数据持久化，数据持久化意味着要写硬盘，在固态硬盘还没有成为标配的今天，硬盘的IO速度永远是系统的痛点。当然了可以在另外节点的内存上进行备份，但是这涉及到了集群的两个稀缺资源：内存和网络。如果因为备份而占用了大量的网络带宽的话，那必将影响系统的性能，吞吐量。

当然了，可以使用日志的方式。但是日志的话对于错误恢复的时间又是不太能接受的。流式计算系统的特点就是要快，如果错误恢复时间太长，那么可能不如直接replay来的快，而且系统设计还更为简单。

其实如果不是为了追求100%的数据丢失，可以使用checkpoint的机制，允许一个时间窗口内的数据丢失。

回到系统设计本身，实际上流式计算系统主要是为了离线和近线的机器学习和数据挖掘，因此肯定要保证数据的处理速度：至少系统可以处理一天的新增数据，否则数据堆积越来越大。因此即使有的数据处理丢失了数据，可以让源头重新发送数据。

还有另外一个话题，就是系统的元数据信心如何保存，因为系统的路由信息等需要是全局可见的，需要保存类似的这些数据以供集群查询。当然了Master节点保持了和所有节点的心跳，它完全可以保存这些数据，并且在心跳中可以返回这些数据。实际上HDFS的NameNode就是这么做的。HDFS的NN这种设计非常合理，为什么这么说？HDFS的元数据包含了非常多的数据：

 

1. 目录文件树结构和文件与数据块的对应关系：会持久化到物理存储中，文件名叫做fsimage。
2. DN与数据块的对应关系，即数据块存储在哪些DN中：在DN启动时会上报到NN它所维护的数据块。这个是动态建立的，不会持久化。因此，集群的启动可能需要比较长的时间。

 

那么对于流式计算系统这种算得上轻量级的元数据来说，Master处理这些元数据实际上要简单的多，当然了，Master需要实现服务的HA和数据的HA。这些不是一个轻松的事情。实际上，可以采用ZooKeeper来保存系统的元数据。ZooKeeper使用一个目录树的结构来保存集群的元数据。节点可以监控感兴趣的数据，如果数据有变化，那么节点会收到通知，然后就保证了系统级别的数据一致性。这点对于系统比较重要，因为节点都是不稳定的，因此系统的其他服务可能都会因为节点失效而发生变化，这些都需要通知相关的节点更新器服务列表，保证了部分节点的失效并不会影响系统的整体的服务，从而也就实现了故障对于用户的透明性。

如何与公司已有的生产环境进行融合

包括Spark和Storm，在国内著名的互联网公司比如百度，淘宝和阿里巴巴都有应用，但是它究竟贡献了多少流量是不得而知的。我了解到的是实际上大部分的流量，尤其是核心流量还是走公司的老架构的。著名的博主陈皓在微博上关于闭源软件和开源软件“特点”之争算是引起了轩然大波，具体讨论可以见知乎。之所以引用这个争论也是为了切合本小节的主题：如何与公司已有的生产环境进行融合。

虽然互联网公司的产品迭代很快，但是公司的核心算法和架构基本上改动不会那么多，因此公司不可能为了推动Storm和Spark这种开源产品而进行大规模的重新开发。只有那么后起的项目，从零开始的项目，比如小规模的调研项目才可能用这些产品。当然了开源产品首先是一个通用的平台，但是通用有可能产生的代价就是不那么高效，对于某些特殊地方的不能根据特殊的应用场景进行优化。如果对这个开源平台进行二次开发，使得性能方面满足自己的需求，首先不管法务上的问题，对于自己私有版本和社区版本进行merge也是个很大的challenge。就像现在很多公司对于Linux进行了二次裁剪，开发自己需要的Linux一样。都需要一些对于这些架构非常熟悉，并且非常熟悉社区动态的人去做这些事情。而这些在互联网公司，基本上是不可能的。因此大部分时候，都是自己做一个系统，去非常高效切合的去满足自身的需求。

当然了，开源社区的闪光点也会影响到闭源产品，闭源产品也会影响开源产品，这个相互影响是良性的，可以推动技术向前发展。

总结

Storm和Spark的设计，绝对不是一篇文章所能解决的。它里边由非常多的哲学需要我们仔细去学习。它们可以说是我们进行系统设计的良好的范例。本博客在接下来的半年会通过Spark的源码来学习Spark的系统架构。敬请期待！

原文链接： 从Storm和Spark Streaming学习流式实时分布式计算系统的设计要点（责编/魏伟）