学习Storm、Spark Streaming流式实时分布式计算系统的设计要点

阅读导读： 1.流式实时分布式计算系统有哪些共同特征，产生的背景是什么？ 2.原语设计的有哪些要点？ 3.元语设计中Spark、storm是如何设计的？ 4.Storm有哪六种消息分发模式？

流式实时分布式计算系统在互联网公司占有举足轻重的地位，尤其在在线和近线的海量数据处理上。在线系统负责处理在线请求，因此低延时高可靠是核心指标。在线系统是互联网公司的核心，系统的好坏直接影响了流量，而流量对互联网公司来说意味着一切。在线系统使用的数据是来自于后台的计算系统产生的。 对于在线（区别于响应互联网用户请求的在线系统，这个在线系统主要是内部使用的，也就是说并不直接服务于互联网用户）/近线系统来说，处理的是线上产生的数据，比如在线系统产生的日志，记录用户行为的数据库等，因此近线系统也需要低延时高可靠的处理海量数据。对于那些时效性很强的数据，比如新闻热点，电商的促销，微博热词等都需要在很短的时间内完成数据处理以供在线系统使用。 而处理这些海量数据的，就是实时流式计算系统。Spark是实时计算的系统，支持流式计算，批处理和实时查询。它使用一个通用的stack解决了很多问题，毕竟任何公司都想要Unified的平台去处理遇到的问题，可以减少开发和维护的人力成本和部署平台的物力成本。除了Spark，流式计算系统最有名的就是Twitter的Storm和Yahoo的S4（其实Spark的流式计算还是要弱于Storm的，个人认为互联网公司对于Storm的部署还是多于Spark的）。 1.流式计算的背景和特点 现在很多公司每天都会产生数以TB级的大数据，如何对这些数据进行挖掘，分析成了很重要的课题。比如： 电子商务：需要处理并且挖掘用户行为产生的数据，产生推荐，从而带来更多的流量和收益。最理想的推荐就是根据兴趣推荐给用户本来不需要的东西！而每天处理海量的用户数据，需要一个低延时高可靠的实时流式分布式计算系统。 新闻聚合：新闻时效性非常重要，如果在一个重大事情发生后能够实时的推荐给用户，那么肯定能增大用户粘性，带来可观的流量。 社交网站：大家每天都会去社交网站是为了看看现在发生了什么，周围人在做什么。流式计算可以把用户关注的热点聚合，实时反馈给用户，从而达到一个圈子的聚合效果。 交通监管部门：每个城市的交通监管部门每天都要产生海量的视频数据，这些视频数据也是以流的形式源源不断的输系统中。实时流式计算系统需要以最快的速度来处理这些数据。 数据挖掘和机器学习：它们实际上是互联网公司内部使用的系统，主要为线上服务提供数据支撑。它们可以说是互联网公司的最核心的平台之一。系统的效率是挖掘的关键，理想条件下就是每天产生的海量数据都能得到有效处理，对于原来的数据进行全量更新。 大型集群的监控：自动化运维很重要，集群监控的实时预警机制也非常重要，而流式系统对于日志的实时处理，往往是监控系统的关键。 等等。 流式实时分布式计算系统就是要解决上述问题的。这些系统的共同特征是什么？ 非常方便的运行用户编写的计算逻辑：就如Hadoop定义了Map和Reduce的原语一样，这些系统也需要让用户关注与数据处理的具体逻辑上，他们不应该也不需要去了解这些usder defined codes是如何在分布式系统上运转起来的。因为他们仅仅关注与数据处理的逻辑，因此可以极大的提高效率。而且应该尽量不要限制编程语言，毕竟不同的公司甚至同一公司的不同部门使用的语言可能是千差万别的。支持多语言无疑可以抢占更多的用户。 Scale-out的设计：分布式系统天生就是scale-out的。 无数据丢失：系统需要保证无数据丢失，这也是系统高可用性的保证。系统为了无数据丢失，需要在数据处理失败的时候选择另外的执行路径进行replay（系统不是简单的重新提交运算，而是重新执行调度，否则按照来源的call stack有可能使得系统永远都在相同的地方出同样的错误）。 容错透明：用户不会也不需要关心容错。系统会自动处理容错，调度并且管理资源，而这些行为对于运行于其上的应用来说都是透明的。 数据持久化：为了保证高可用性和无数据丢失，数据持久化是无法躲避的问题。的确，数据持久化可能在低延时的系统中比较影响性能，但是这无法避免。当然了，如果考虑到出错情况比较少，在出错的时候我们能够忍受数据可以从头replay，那么中间的运算可以不进行持久化。注意，这只有在持久化的成本要比计算的replay高的情况下有效。一般来说，计算的结果需要replica，当然了，可以使用将数据replica到其他的节点的内存中去（这又会占用集群的网络带宽）。 超时设置：超时之所以在在这里被提出来，因为超时时间的大小设置需要重视，如果太短可以会误杀正常运行的计算，如果太长则不能快速的检测错误。还有就是对于错误的快速发现可以这类系统的一个设计要点，毕竟，超时了才发现错误很多时候在时效性上是不可接受的。 2.原语设计 Hadoop定义了Map和Reduce，使得应用者只需要实现MR就可以实现数据处理。而流式系统的特点，允许它们可以进行更加具体一些的原语设计。流式的数据的特点就是数据时源源不断进入系统的，而这些数据的处理一般都需要几个阶段。拿普通的日志处理来说，我们可能仅仅关注Error的日志，那么系统的第一个计算逻辑就是进行filer。接下来可能需要对这个日志进行分段，分段后可能交给不同的规则处理器进行处理。因此，数据处理一般是分阶段的，可以说是一个有向无环图，或者说是一个拓扑。实际上，Spark抽象出的运算逻辑就是由RDD（Resilient Distributed Dataset）构成DAG（Directed Acyclic Graph），而Storm则有Spout和Blot构成Topology（拓扑）。 Spark的设计 原来实现Spark流处理的是Shark，现在Shark已经停止更新了。而最新的Spark的流处理都交给Spark Streaming了。Spark Streaming是将流式计算分解成一系列短小的批处理作业。这里的批处理引擎是Spark，也就是把Spark Streaming的输入数据按照batch size（如1秒）分成一段一段的数据，每一段数据都转换成Spark中的RDD，然后将Spark Streaming中对DStream的Transformation操作变为针对Spark中对RDD的Transformation操作，将RDD经过操作变成中间结果保存在内存中。 整个流式计算根据业务的需求可以对中间的结果进行叠加，或者存储到外部设备。下图显示了Spark Streaming的整个流程。

WordCount的例子：

1. // Create the context and set up a network input stream to receive from a host:port
2. val ssc = new StreamingContext(args(0), "NetworkWordCount", Seconds(1))
3. val lines = ssc.socketTextStream(args(1), args(2).toInt)
5. // Split the lines into words, count them, and print some of the counts on the master
6. val words = lines.flatMap(_.split(" "))
7. val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
8. wordCounts.print()
10. // Start the computation
11. ssc.start()

这个例子使用Scala写的，一个简单优雅的函数式编程语言，同时也是基于JVM的后Java类语言。 Storm的设计 Storm将计算逻辑成为Topology，其中Spout是Topology的数据源，这个数据源可能是文件系统的某个日志，也可能是MessageQueue的某个消息队列，也有可能是数据库的某个表等等；Bolt负责数据的护理。Bolt有可能由另外两个Bolt的join而来。 而Storm最核心的抽象Streaming就是连接Spout，Bolt以及Bolt与Bolt之间的数据流。而数据流的组成单位就是Tuple（元组），这个Tuple可能由多个Fields构成，每个Field的含义都在Bolt的定义的时候制定。也就是说，对于一个Bolt来说，Tuple的格式是定义好的。

原语设计的要点 流式系统的原语设计，要关注一下几点： 如何定义计算拓扑：要方便算法开发者开发算法与策略。最好的实现是定义一个算法与框架的交互方式，定义好算法的输入结构和算法的输出结构。然后拓扑能够组合不同的算法来为用户提供一个统一的服务。计算平台最大的意义在于算法开发者不需要了解程序的运行，并发的处理，高可用性的实现，只需要提供算法与计算逻辑即可以快速可靠的处理海量的数据。 拓扑的加载与启动：对于每个节点来说，启动时需要加载拓扑，节点需要其他的信息，比如上游的数据来源与下游的数据输出。当然了下游的数据输出的拓扑信息可以存储到Tuple中，对于数据需要放到那里去拓扑本身是无状态的。这就取决于具体的设计了。 拓扑的在线更新：对于每个算法逻辑来说，更新是不可避免的，如何在不停止服务的情况下进行更新是必要的。由于实现了架构与算法的剥离，因此算法可以以一个单独的个体进行更新。可以操作如下：Master将算法实体保存到一个Worker可见的地方，比如HDFS或者是NFS或者ZK，然后通过心跳发送命令到拓扑，拓扑会暂时停止处理数据而加载新的算法实体，加载之后重新开始处理数据。数据一般都会放到buffer中，这个buffer可能是一个queue。但是从外界看来，拓扑实际上是一直处于服务状态的。 数据如何流动：流式系统最重要的抽象就是Streaming了。那么Steaming如何流动？实际上涉及到消息的传递和分发，数据如何从一个节点传递到另外一个节点，这是拓扑定义。 计算的终点及结果处理：流式计算的特点就是计算一直在进行，流是源源不断的流入到系统中的。但是对于每个数据单位来说它的处理结果是确定的，这个结果一般是需要返回调用者或者需要持久化的。比如处理一个时间段的交通违章，那么输入的数据是一段时间的视频监控，输出这是违章的信息，比如车牌，还有违章时刻的抓拍的图片。这个数据要么返回调用者，由调用者负责数据的处理，包括持久化等。或者是拓扑最后的节点将这些信息进行持久化。系统需要对这些常见的case进行指导性的说明，需要在Programmer Guide的sample中给出使用例子。 3.消息传递和分发 对于实现的逻辑来说，它们都是有向无环图的一个节点，那么如何设计它们之间的消息传递呢？或者说数据如何流动的？因为对于分布式系统来说，我们不能假定整个运算都是在同一个节点上（事实上，对于闭源软件来说，这是可以的，比如就是满足一个特定运算下的计算，计算平台也不需要做的那么通用，那么对于一个运算逻辑让他在一个节点完成也是可以了，毕竟节省了调度和网络传输的开销）。或者说，对于一个通用的计算平台来说，我们不能假定任何事情。 消息传递和分发是取决于系统的具体实现的。通过对比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有六种消息分发模式： Shuffle Grouping: 随机分组，Storm会尽量把数据平均分发到下游Bolt中。 Fields Grouping：按字段分组， 比如按userid来分组， 具有同样userid的tuple会被分到相同的Bolt。这个对于类似于WordCount这种应用非常有帮助。 All Grouping： 广播， 对于每一个Tuple， 所有的Bolts都会收到。这种分发模式要慎用，会造成资源的极大浪费。 Global Grouping: 全局分组， 这个Tuple被分配到storm中的一个bolt的其中一个task。这个对于实现事务性的Topology非常有用。 Non Grouping: 不分组， 这个分组的意思是说stream不关心到底谁会收到它的tuple。目前这种分组和Shuffle grouping是一样的效果， 有一点不同的是storm会把这个bolt放到这个bolt的订阅者同一个线程里面去执行。 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有以下缺点： 不容易部署，尤其是在云环境下：以为ZMQ是以Ｃ写的，因此它还是紧依赖于操作系统环境的。 无法限制其内存。通过JVM可以很容易的限制java所占用的内存。但是ZMQ对于Storm来说是个黑盒似得存在。 Storm无法从ZMQ获取信息。比如Storm无法知道当前buffer中有多少数据为发送。 当然了还有所谓的性能问题，具体可以访问Netty作者的blog。结论就是Netty的性能比ZMQ（在默认配置下）好两倍。不知道所谓的ZMQ的默认配置是什么。反正我对这个结果挺惊讶。当然了，Netty使用Java实现的确方便了在Worker之间的通信加上授权和认证机制。这个使用ZMQ的确是不太好做。 4.高可用性 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所能涵盖的了。 5.存储模型与数据不丢失 其实，数据不丢失有时候和处理速度是矛盾的。为了数据不丢失就要进行数据持久化，数据持久化意味着要写硬盘，在固态硬盘还没有成为标配的今天，硬盘的IO速度永远是系统的痛点。当然了可以在另外节点的内存上进行备份，但是这涉及到了集群的两个稀缺资源：内存和网络。如果因为备份而占用了大量的网络带宽的话，那必将影响系统的性能，吞吐量。 当然了，可以使用日志的方式。但是日志的话对于错误恢复的时间又是不太能接受的。流式计算系统的特点就是要快，如果错误恢复时间太长，那么可能不如直接replay来的快，而且系统设计还更为简单。其实如果不是为了追求100%的数据丢失，可以使用checkpoint的机制，允许一个时间窗口内的数据丢失。回到系统设计本身，实际上流式计算系统主要是为了离线和近线的机器学习和数据挖掘，因此肯定要保证数据的处理速度： 至少系统可以处理一天的新增数据，否则数据堆积越来越大。因此即使有的数据处理丢失了数据，可以让源头重新发送数据。 还有另外一个话题，就是系统的元数据信心如何保存，因为系统的路由信息等需要是全局可见的，需要保存类似的这些数据以供集群查询。当然了Master节点保持了和所有节点的心跳，它完全可以保存这些数据，并且在心跳中可以返回这些数据。 实际上HDFS的NameNode就是这么做的。HDFS的NN这种设计非常合理，为什么这么说？HDFS的元数据包含了非常多的数据：目录文件树结构和文件与数据块的对应关系：会持久化到物理存储中，文件名叫做fsimage。 DN与数据块的对应关系，即数据块存储在哪些DN中：在DN启动时会上报到NN它所维护的数据块。这个是动态建立的，不会持久化。因此，集群的启动可能需要比较长的时间。 那么对于流式计算系统这种算得上轻量级的元数据来说，Master处理这些元数据实际上要简单的多，当然了，Master需要实现服务的HA和数据的HA。这些不是一个轻松的事情。实际上，可以采用ZooKeeper来保存系统的元数据。ZooKeeper使用一个目录树的结构来保存集群的元数据。节点可以监控感兴趣的数据，如果数据有变化，那么节点会收到通知，然后就保证了系统级别的数据一致性。这点对于系统比较重要，因为节点都是不稳定的，因此系统的其他服务可能都会因为节点失效而发生变化，这些都需要通知相关的节点更新器服务列表，保证了部分节点的失效并不会影响系统的整体的服务，从而也就实现了故障对于用户的透明性。 6.如何与公司已有的生产环境进行融合 包括Spark和Storm，在国内著名的互联网公司比如百度，淘宝和阿里巴巴都有应用，但是它究竟贡献了多少流量是不得而知的。我了解到的是实际上大部分的流量，尤其是核心流量还是走公司的老架构的。著名的博主陈皓在微博上关于闭源软件和开源软件“特点”之争算是引起了轩然大波，具体讨论可以见知乎。之所以引用这个争论也是为了切合本小节的主题：如何与公司已有的生产环境进行融合。 虽然互联网公司的产品迭代很快，但是公司的核心算法和架构基本上改动不会那么多，因此公司不可能为了推动Storm和Spark这种开源产品而进行大规模的重新开发。只有那么后起的项目，从零开始的项目，比如小规模的调研项目才可能用这些产品。当然了开源产品首先是一个通用的平台，但是通用有可能产生的代价就是不那么高效，对于某些特殊地方的不能根据特殊的应用场景进行优化。如果对这个开源平台进行二次开发，使得性能方面满足自己的需求，首先不管法务上的问题，对于自己私有版本和社区版本进行merge也是个很大的challenge。就像现在很多公司对于Linux进行了二次裁剪，开发自己需要的Linux一样。都需要一些对于这些架构非常熟悉，并且非常熟悉社区动态的人去做这些事情。而这些在互联网公司，基本上是不可能的。因此大部分时候，都是自己做一个系统，去非常高效切合的去满足自身的需求。 当然了，开源社区的闪光点也会影响到闭源产品，闭源产品也会影响开源产品，这个相互影响是良性的，可以推动技术向前发展。