分布式流式处理框架：Storm

Storm是一个免费开源、分布式、高容错的实时计算系统。它与其他大数据解决方案的不同之处在于它的处理方式。Hadoop 在本质上是一个批处理系统，数据被引入 Hadoop 文件系统 (HDFS) 并分发到各个节点进行处理。当处理完成时，结果数据返回到 HDFS 供始发者使用。Hadoop的高吞吐，海量数据处理的能力使得人们可以方便地处理海量数据。但是，Hadoop的缺点也和它的优点同样鲜明——延迟大，响应缓慢，运维复杂。Storm就是为了弥补Hadoop的实时性为目标而被创造出来。Storm 支持创建拓扑结构来转换没有终点的数据流。不同于 Hadoop 作业，这些转换从不停止，它们会持续处理到达的数据。Storm经常用于在实时分析、在线机器学习、持续计算、分布式远程调用和ETL等领域。Storm的部署管理非常简单，而且，在同类的流式计算工具，Storm的性能也是非常出众的。

Strom的优点：

• 简单的编程模型。类似于MapReduce降低了并行批处理复杂性，Storm降低了进行实时处理的复杂性。
• 可以使用各种编程语言。你可以在Storm之上使用各种编程语言。默认支持Clojure、Java、Ruby和Python。要增加对其他语言的支持，只需实现一个简单的Storm通信协议即可。
• 容错性。Storm会管理工作进程和节点的故障。模块都是无状态的，随时宕机重启。由于是分布式，一个节点挂了不能影响系统的正常运行。
• 水平扩展。计算是在多个线程、进程和服务器之间并行进行的。
• 可靠的消息处理。Storm保证每个消息至少能得到一次完整处理。任务失败时，它会负责从消息源重试消息。
• 快速。系统的设计保证了消息能得到快速的处理，使用ZeroMQ（新的消息机制使用netty代替ZeroMQ）作为其底层消息队列。
• 本地模式。Storm有一个“本地模式”，可以在处理过程中完全模拟Storm集群。这让你可以快速进行开发和单元测试。

Storm的组成：

在介绍Storm前我们先来看下它与Hadoop的对比：

Storm主要分为两种组件Nimbus和Supervisor。这两种组件都是快速失败的，没有状态。任务状态和心跳信息等都保存在Zookeeper上的，提交的代码资源都在本地机器的硬盘上。Storm中的一些概念：

• Nimbus：负责资源分配和任务调度。集群里面发送代码，分配工作给机器，并且监控状态。全局只有一个。
• Supervisor：负责接受nimbus分配的任务，启动和停止属于自己管理的worker进程。会监听分配给它那台机器的工作，根据需要启动/关闭工作进程Worker。每一个要运行Storm的机器上都要部署一个，并且，按照机器的配置设定上面分配的槽位数。
• Worker：运行具体处理组件逻辑的进程。
• Task：worker中每一个spout/bolt的线程称为一个task. 在storm0.8之后，task不再与物理线程对应，同一个spout/bolt的task可能会共享一个物理线程，该线程称为executor。
• Zookeeper：Storm重点依赖的外部资源。Nimbus和Supervisor甚至实际运行的Worker都是把心跳保存在Zookeeper上的。Nimbus也是根据Zookeerper上的心跳和任务运行状况，进行调度和任务分配的。
• Topology：storm中运行的一个实时应用程序，因为各个组件间的消息流动形成逻辑上的一个拓扑结构。Topology处理的最小的消息单位是一个Tuple，也就是一个任意对象的数组。Topology由Spout和Bolt构成。
• Spout：在一个topology中产生源数据流的组件。通常情况下spout会从外部数据源（如Message Queue、RDBMS、NoSQL、Realtime Log）中读取数据，然后转换为topology内部的源数据。Spout是一个主动的角色，其接口中有个nextTuple()函数，storm框架会不停地调用此函数，用户只要在其中生成源数据即可。
• Bolt：在一个topology中接受数据然后执行处理的组件。Bolt可以执行过滤、函数操作、合并、写数据库等任何操作。Bolt是一个被动的角色，其接口中有个execute(Tuple input)函数,在接受到消息后会调用此函数，用户可以在其中执行自己想要的操作。
• Tuple：一次消息传递的基本单元。本来应该是一个key-value的map，但是由于各个组件间传递的tuple的字段名称已经事先定义好，所以tuple中只要按序填入各个value就行了，所以就是一个value list。
• Stream：源源不断传递的tuple就组成了stream。是Storm中对数据进行的抽象，它是时间上无界的tuple元组序列。在Topology中，Spout是Stream的源头，负责为Topology从特定数据源发射Stream；Bolt可以接收任意多个Stream作为输入，然后进行数据的加工处理过程，如果需要，Bolt还可以发射出新的Stream给下级Bolt进行处理。
• Stream Grouping即消息的partition方法。目前Storm中提供了以下7种Stream Grouping策略：Shuffle Grouping、Fields Grouping、All Grouping、Global Grouping、Non Grouping、Direct Grouping、Local or shuffle grouping，具体策略可以参考 这里。

下图描述了Nimbus、Supervisor、Worker、Task、Zookeeper这几个角色之间的关系：

在Storm中，一个实时应用的计算任务被打包作为Topology发布，这同Hadoop的MapReduce任务相似。但是有一点不同的是：在Hadoop中，MapReduce任务最终会执行完成后结束；而在Storm中，Topology任务一旦提交后永远不会结束，除非你显示去停止任务。计算任务Topology是由不同的Spouts和Bolts，通过数据流（Stream）连接起来的图。下面是一个Topology的结构示意图：

Topology中每一个计算组件（Spout和Bolt）都有一个并行执行度，在创建Topology时可以进行指定，Storm会在集群内分配对应并行度个数的线程来同时执行这一组件。既然对于一个Spout或Bolt，都会有多个task线程来运行，那么如何在两个组件（Spout和Bolt）之间发送tuple元组呢？Storm提供了若干种数据流分发（Stream Grouping）策略用来解决这一问题。在Topology定义时，需要为每个Bolt指定接收什么样的Stream作为其输入（注：Spout并不需要接收Stream，只会发射Stream）。

下图是Topology的提交流程图：

Storm 的一个最有趣的地方是它注重容错和管理。Storm 实现了有保障的消息处理，所以每个元组都会通过该拓扑结构进行全面处理；如果发现一个元组还未处理，它会自动从喷嘴处重放。Storm 还实现了任务级的故障检测，在一个任务发生故障时，消息会自动重新分配以快速重新开始处理。Storm 包含比 Hadoop 更智能的处理管理，流程会由监管员来进行管理，以确保资源得到充分使用。

下图是Storm的数据交互图。可以看出两个模块Nimbus和Supervisor之间没有直接交互。状态都是保存在Zookeeper上。Worker之间通过ZeroMQ（新的消息机制使用netty代替ZeroMQ）传送数据。

Storm 使用 ZeroMQ 传送消息，这就消除了中间的排队过程，使得消息能够直接在任务自身之间流动。在消息的背后，是一种用于序列化和反序列化 Storm 的原语类型的自动化且高效的机制。

Storm的应用：

Storm被广泛应用于实时分析，在线机器学习，持续计算、分布式远程调用等领域。如果，业务场景中需要低延迟的响应，希望在秒级或者毫秒级完成分析、并得到响应，而且希望能够随着数据量的增大而拓展。那就可以考虑使用Storm。Storm的适用场景：

• 流数据处理。Storm可以用来处理源源不断流进来的消息，处理之后将结果写入到某个存储中去。
• 分布式rpc。由于storm的处理组件是分布式的，而且处理延迟极低，所以可以作为一个通用的分布式rpc框架来使用。

来看一些实际的应用:

• 一淘- 实时分析系统pora：实时分析用户的属性，并反馈给搜索引擎。最初，用户属性分析是通过每天在云梯上定时运行的MR job来完成的。为了满足实时性的要求，希望能够实时分析用户的行为日志，将最新的用户属性反馈给搜索引擎，能够为用户展现最贴近其当前需求的结果。
• 携程- 网站性能监控：实时分析系统监控携程网的网站性能。利用HTML5提供的performance标准获得可用的指标，并记录日志。Storm集群实时分析日志和入库。使用DRPC聚合成报表，通过历史数据对比等判断规则，触发预警事件。

参考资料：

• https://github.com/nathanmarz/storm
• http://xumingming.sinaapp.com/about/