Storm-实时计算系统

作者 | WenasWei

一 Storm

1.1 Storm简介

Storm 最早是由 BackType 公司开发的实时处理系统，底层由 Clojure 实现。Clojure 也是一门基于 JVM 的高级面向函数式的编程语言。2011年 Twitter 公司收购了 BackType 公司，便使用 Storm 帮助企业解决了实时海量数据处理的问题。阿里巴巴在 Storm 基础上，使用 Java 代替 Clojure 实现了核心，并在性能上进行了优化，产生了JStorm。目前 Storm 和 JStorm 都由 Apache 基金会组织管理。

Storm-实时计算系统:

1.2 Storm特性

简单的编程模型: 类似于 MapReduce 降低了并行批处理复杂性，Storm降低了进行实时处理的复杂性，主要通过 Tuple 元组在各个组件之间传递数据。
支持多种编程语言: 默认支持 Clojure、Java、Ruby 和 Python，要增加对其他语言的支持，只需实现一个简单的 Storm 通信协议即可
容错性: Storm 会管理工作进程和节点的故障
水平扩展: 计算是在多个线程、进程和服务器之间并行进行的
可靠的消息处理: Storm 提供了 Ack 机制，保证每个消息至少能得到一次完整处理。任务失败时, 它会负责从消息源重试消息
高性能与低延迟: 系统的设计保证了消息能得到快速的处理，使用 ØMQ¹ 作为其底层消息队列
本地模式: Storm 有一个“本地模式”, 可以在处理过程中完全模拟 Storm 集群

1.3 Storm使用场景

Apache Storm 有许多用例：实时分析，在线机器学习，连续计算，分布式 RPC，ETL 等。Apache Storm 速度很快：基准测试表明它每秒可处理每个节点超过一百万个元组。它具有可扩展性，容错性，可确保您的数据将得到处理，并且易于设置和操作。

（1）信息流处理 Stream Processing
处理源源不断产生的消息，并将处理之后的结果存储到某个存储系统中去，典型的架构是 Kafka + Storm + HBase（或redis）。
（2）连续计算 Continuous Computation
Storm可进行连续查询并把结果即时反馈给客户端，比如把微博上的热门话题发送到浏览器上。
（3）分布式远程过程调用 Distributed RPC
用来处理并行密集查询，客户端向Storm提交一个查询请求和查询参数，Storm 运行 Topology 并行处理，并将结果同步返回给客户端。
（4）在线机器学习。
（5）日志分析: 在线实时分析业务系统或者网站产生的日志数据。
（6）管道传输: 能够使数据在不同系统之间进行传输。
（7）统计分析: 实时获取日志或者消息，对待定得到字段进行统计计数或累加计算。

二 Storm的体系结构

2.1 Storm的核心组件

Storm 的核心组件分为:

Topology: 计算拓扑
Nimbus: Storm 的 Master
Supervisor: Storm 的 Slave
Worker: 工作进程
Task: 运行于Spout或Bolt中的线程
Stream: 数据流
Spout: 数据源
Bolt: 处理数据
Reliability: 可靠性
Tuple: 消息传递的基本单元
Executor: Worker进程中的具体的物理线程
Stream grouping: 每个 Bolt 的确定输入数据流

1、Topology

计算拓扑, Storm 的拓扑是对实时计算应用逻辑的封装，它的作用与 MapReduce 的任务（Job）很相似，区别在于 MapReduce 的一个 Job 在得到结果之后总会结束，而拓扑会一直在集群中运行，直到你手动去终止它；拓扑还可以理解成由一系列通过数据流（Stream Grouping）相互关联的 Spout 和 Bolt 组成的的拓扑结构, 一个完整的运行任务被称为一个 Topology；监听集群状态，当 Supervisor 节点挂掉，由 Nimbus 将该节点上正在运行着的任务重新分配给其他 Supervisor 执行。

2、Nimbus

即 Storm 的 Master，本身无状态，负责资源分配和任务调度；一个 Storm 集群只有一个 Nimbus，分配工作给从节点 supervisor，注意不是直接分配，而是将任务发布到 zookeeper 上，由 supervisor 到 zookeeper上领取任务。

3、Supervisor

即 Storm 的 Slave，负责接收 Nimbus 分配的任务，管理所有 Worker，一个 Supervisor 节点中包含多个 Worker 进程。

4、Worker

工作进程，每个工作进程中都有多个 Task。

5、Task

任务，Storm中最小的处理单元，在 Storm 集群中每个 Spout 和 Bolt 都由若干个任务（tasks）来执行；每个任务都与一个执行线程相对应。

6、Stream

数据流（Streams）是 Storm 中最核心的抽象概念。一个数据流指的是在分布式环境中并行创建、处理的一组元组（tuple）的无界序列。

7、Spout

数据源（Spout）是拓扑中数据流的来源。一般 Spout 会从一个外部的数据源读取元组然后将他们发送到拓扑中。根据需求的不同，Spout 既可以定义为可靠的数据源，也可以定义为不可靠的数据源。一个可靠的 - Spout 能够在它发送的元组处理失败时重新发送该元组，以确保所有的元组都能得到正确的处理；相对应的，不可靠的 Spout 就不会在元组发送之后对元组进行任何其他的处理。一个 Spout 可以发送多个数据流。

8、Bolt

拓扑中所有的数据处理均是由 Bolt 完成的。通过数据过滤（filtering）、函数处理（functions）、聚合（aggregations）、联结（joins）、数据库交互等功能，Bolt 几乎能够完成任何一种数据处理需求。一个 Bolt 可以实现简单的数据流转换，而更复杂的数据流变换通常需要使用多个 Bolt 并通过多个步骤完成。

9、Reliability

可靠性。Storm 可以通过拓扑来确保每个发送的元组都能得到正确处理；通过跟踪由 Spout 发出的每个元组构成的元组树可以确定元组是否已经完成处理。每个拓扑都有一个“消息延时”参数，如果 Storm 在延时时间内没有检测到元组是否处理完成，就会将该元组标记为处理失败，并会在稍后重新发送该元组。

10、Tuple

消息传递的基本单元

11、Executor

Storm0.8之后, Executor为 Worker进程中的具体的物理线程,同一个 Spout、Bolt的Task可能会共享一个物理线程,一个 Executor中只能运行隶属于同一个 Spout/ Bolt 的 Task

12、Stream grouping

Stream grouping 为拓扑中的每个 Bolt 的确定输入数据流是定义一个拓扑的重要环节；数据流分组定义了在 Bolt 的不同任务（tasks）中划分数据流的方式。

2.2 Storm集群架构

Storm 的集群也是采用主从架构，Storm 的 Topology 相比于 Hadoop 中的 MapReduceJob 关键区别是: 一个 Topology 在启动后，会永远运行; 而一个 MapReduceJob 会有启动、运行到最终会结束的过程。

在 Storm 的集群里面有两种节点: 控制节点 (Nimbus) 和工作节点(Supervisor), Nimbus 负责在集群里面分发执行代码，分配工作给工作节点，并且监控任务的执行状态。每一个工作节点上面运行一个叫作 Supervisor 的守护进程。Supervisor 会监听分配给自己所在机器的工作，根据需要启动/关闭工作进程。每一个工作进程执行一个 Topology 的一个子集，一个运行的 Topology 由运行在很多机器上的很多工作进程组成。

Storm 集群可以使用 Apache Zookeeper 集群来作为自己的协调系统，Storm 集群的组织架构:

Nimbus 节点和 Supervisor 节点之间的数据通信和协调依靠 Zookeeper 框架完成， Storm被设计成无状态的，Nimbus 节点和 Supervisor 节点中不会保存任何状态信息，所有的状态信息会被保存在 Zookeeper 或者磁盘中。这样，当 Nimbus 节点或者 Supervisor 节点宕机，或者直接使用 kill -9命令杀死 Nimbus 进程和 Supervisor 进程时，重启就可以继续工作，使 Storm 具有高度的稳定性和可靠性。

Storm任务分配流程图:

(1) 客户端提交拓扑到 nimbus；
(2) Nimbus 针对该拓扑建立本地的目录根据 Topology 的配置计算 Task，分配 Task，在 Zookeeper 上建立 Assignments 节点存储 Task 和 Supervisor 机器节点中 Woker 的对应关系；
(3) 在 Zookeeper 上创建 Task beats 节点来监控 Task 的心跳；启动 topology。
(4) Supervisor 去 Zookeeper 上获取分配的 Tasks，启动多个 Woker 进行，每个 Woker 启动一个或者多个 Executors，线程中执行 Task；
(5) 根据 Topology 信息初始化建立 Task 之间的连接; Task 和 Task 之间是通过 ZeroMQ、NettyRPC 管理的通信；

三 Stream grouping的9种分组策略

在 Storm 中有八种内置的数据流分组方式，还有一种自定义分组策略。

3.1 随机分组（Shuffle grouping）

这种方式下元组会被尽可能随机地分配到 Bolt 的不同任务（tasks）中，使得每个任务所处理元组数量能够能够保持基本一致，以确保集群的负载均衡。

Shuffle Grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("mySpout",new MySpout(),1); builder.setBolt("myBolt",new myBolt(),2).shuffleGrouping("mySpout");

3.2 域分组（Fields grouping）

数据流根据定义的“域”来进行分组, 按 Tuple 中的字段分组，相同字段的 Tuple 会被分发到同一个 Task 中，这样便于数据的跟踪和处理。例如，如果某个数据流是基于一个名为 “user-id” 的域进行分组的，那么所有包含相同的 “user-id” 的元组都会被分配到同一个任务中，这样就可以确保消息处理的一致性。

Fields grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder();builder.setSpout("mySpout",new MySpout(),1);// 按照 myFieId 字段进行分区builder.setBolt("myBolt",new myBolt(),2).fieldsGrouping("mySpout", new Fields("myFieId")).fieldsGrouping(ConstData.CACHE_BOLT, ConstData.PROCESSOR_STREAM, new Fields(AlarmAttrName.ME));

3.3 部分关键字分组（Partial Key grouping）

这种方式与域分组很相似，根据定义的域来对数据流进行分组，不同的是，这种方式会考虑下游 Bolt 数据处理的均衡性问题，在输入数据源关键字不平衡时会有更好的性能1。感兴趣的读者可以参考这篇论文，其中详细解释了这种分组方式的工作原理以及它的优点。

Partial Key grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder();builder.setSpout("spout", new PartialKeyGroupingSpout(), 2).setNumTasks(3);builder.setBolt("bolt", new PartialKeyGroupingBolt(), 3).setNumTasks(5).partialKeyGrouping("spout", new Fields("flag"));

3.4 完全分组（All grouping）

数据流会被同时发送到 Bolt 的所有任务中（同一个元组会被复制多份然后被所有的任务处理）, 也就是说所有的 Tuple 都会被广播发送到所有的 Bolt，常被用于向 Bolt 发送的信号，使用这种分组方式要特别小心。

All grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder(); builder.setSpout("mySpout",new MySpout(),1);builder.setBolt("myBolt",new MyBolt(),2).allGrouping("mySpout");

3.5 全局分组（Global grouping）

这种方式下所有的数据流都会被发送到 Bolt 的同一个任务中，也就是 id 最小的那个任务。

Global grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder();// 拓扑名、数据源、并行度builder.setSpout("mySpout", new MySpout(), 1);builder.setBolt("bolt", new MyBolt(), 2).globalGrouping("mySpout");

3.6 非分组（None grouping）

使用这种方式说明你不关心数据流如何分组。目前这种方式的结果与随机分组完全等效，不过未来 Storm 社区可能会考虑通过非分组方式来让 Bolt 和它所订阅的 Spout 或 Bolt 在同一个线程中执行。

None grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder();// 拓扑名、数据源、并行度builder.setSpout("mySpout", new MySpout(), 1);// 等于 shuffleGroupingbuilder.setBolt("myBolt", new MyBolt(), 2).noneGrouping("mySpout");

3.7 直接分组（Direct grouping）

这是一种特殊的分组方式。使用这种方式意味着元组的发送者可以指定下游的哪个任务可以接收这个元组。只有在数据流被声明为直接数据流时才能够使用直接分组方式。使用直接数据流发送元组需要使用 OutputCollector 的其中一个 emitDirect 方法。Bolt 可以通过 TopologyContext 来获取它的下游消费者的任务 id，也可以通过跟踪 OutputCollector 的 emit 方法（该方法会返回它所发送元组的目标任务的 id）的数据来获取任务 id。

Direct grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder();// 设置 Spoutbuilder.setSpout("mySpout", new Myspout()).setNumTasks(2);//设置 creator-Boltbuilder.setBolt("myBolt", new MyBolt(), 2).directGrouping("mySpout").setNumTasks(2);

3.8 本地或随机分组（Local or shuffle grouping）

如果在源组件的 worker 进程里目标 Bolt 有一个或更多的任务线程，元组会被随机分配到那些同进程的任务中。换句话说，这与随机分组的方式具有相似的效果。

Local or shuffle grouping 简单代码示例:

        TopologyBuilder builder = new TopologyBuilder();builder.setSpout("mySpout", new MySpout(), 1);builder.setBolt("myBolt", new myBolt(), 2).local0rShuffleGrouping("mySpout");

3.9 自定义分组策略（Custom Grouping）

当 Storm 内置的7种分组策略无法满足需求时, 还可以使用自定义分组策略:

Custom Grouping

实现自定义分组策略需要实现 CustomStreamGrouping 接口。自定义分组策略类 MyCustomGrouping 的代码如下所示:

public class Mycustomgrouping implements CustomStreamGrouping {@Overridepublic void prepare(WorkerTopologyContext workerTopologyContext, GlobalStreamId globalStreamId, List<Integer> list) {}@Overridepublic List<Integer> chooseTasks(int i, List<Object> list) {return null;}
}

使用示例代码如下所示:

        TopologyBuilder builder = new TopologyBuilder();// 拓扑名、数据源、并行度builder.setSpout("mySpout", new MySpout(), 1);builder.setBolt("myBolt", new MyBolt(), 2).customGrouping("mySpout", new Mycustomgrouping());

四 Storm的可靠性机制和容错性

4.1 Storm的可靠性机制

在 Storm 中运行着一类特殊的 Task，称为 acker 。acker 能够监控每个从 Spout 发送的 tuple 产生的消息树。当 acker 发现一个 tuple 产生的消息树被完整地处理后，会向产生该 tuple 的 Task 发送一条通知消息，表示该消息已经执行完成。用户可以通过配置 Config.TOPOLOGY_ACKERS 的值来设置一个 Topology 中acker 的数量，默认值为1。如果一个 Topology 中 tuple 的数量较多，就可以将 acker 的数量设置得大一些，以此提高整个 Topology 运行时的性能。

Storm Ack 机制的简易流程:

(1) Spout 进行初始化，会产生一个全局唯一的 taskId。
(2) Spout 创建一个 Tuple，并为 Tuple 指定一个 tupleld，在各个组件之间进行数据传输
(3) Spout 将创建的 Tuple 发送出去, 并根据指定的 tupleld 开启跟踪，同时将消息发送到 Acker，Acker对消息进行跟踪。
(4) Spout 发送的 Tuple 会经过一系列 Bolt 的处理，产生一个 Tuple 跟踪列表
(5) Bolt 调用 collectorack() 方法进行一系列的异或操作，如果得出的最终结果是0，则表示成功;如果得出的最终结果不是0，则表示失败。
(6) 根据第(1)步生成的 taskId，回调 Spout 中的 ack()方法或者 fail()方法

注意:

用户如果不关心可靠性的问题，可以使用如下方式进行设置:

(1) 在定义 Topology 时设置 Acker 的数量为 0。代码如下所示:

conf.setNumAckers(0);

(2) 在 Spout 发送 Tuple 时不指定 messageld, Storm不会开启跟踪，这样就不会开启 Acker 确认机制。
(3) 在 Bolt 中使用以 RichBolt 结尾的 Bolt 类，不手动调用 collector.ack() 方法和 collector.fail() 方法，即不进行数据锚定来发送新的 Tuple 数据, 因为以 RichBolt 结尾的 Bolt 类没有提供可靠性机制，而以 BasicBolt 为结尾的 Bolt 封装了可靠性机制。

4.2 Storm的容错性

Storm 的容错分为如下类型:

（1）工作进程 worker 失效

如果一个节点的工作进程 worker “死掉”，supervisor 进程会尝试重启该 worker。如果连续重启 worker 失败或者 worker 不能定期向 Nimbus 报告“心跳”，Nimbus 会分配该任务到集群其他的节点上执行。

（2）集群节点失效

如果集群中某个节点失效，分配给该节点的所有任务会因超时而失败，Nimbus 会将分配给该节点的所有任务重新分配给集群中的其他节点。

（3）Nimbus 或者supervisor 守护进程失败

Nimbus 和supervisor 都被设计成快速失败（遇到未知错误时迅速自我失败）和无状态的（所有的状态信息都保存在 Zookeeper 上或者是磁盘上）。Nimbus 和supervisor 守护进程必须在一些监控工具（例如，daemontools 或者 monitor ）的辅助下运行，一旦 Nimbus 或者 supervisor 失败，可以立刻重启它们，整个集群就好像什么事情也没发生。最重要的是，没有工作进程 worker 会因为Nimbus 或 supervisor 的失败而受到影响，Storm 的这个特性和 Hadoop 形成了鲜明的对比，如果 JobTracker 失效，所有的任务都会失败。

（4）Nimbus 所在的节点失效

如果 Nimbus 守护进程驻留的节点失败，工作节点上的工作进程 worker 会继续执行计算任务，而且，如果 worker 进程失败，supervisor 进程会在该节点上重启失败的worker 任务。但是，没有 Nimbus 的影响时，所有 worker 任务不会分配到其他的工作节点机器上，即使该 worker 所在的机器失效。

五 Storm API

5.1 Storm 组件回顾:

5.2 Storm API

(1) 首先编写数据源类: Spout

a. 继承 BaseRichSpout类
b. 实现 IRichSpout接口
c. 重点需要几个方法进行重写或实现:open、nextTuple 和 DeclareOutputFields 方法

(2) 继续编写数据处理类:Bolt

a. 继承 BaseBasicBolt类
b. 实现 IRichBolt接口
c. 重点需要几个方法进行重写或实现: execute、 declareOutputFields方法

(3) 最后编写主函数( Topology)去进行提交一个任务

a. TopologyBuilder 组件Topology
b. 在使用 Topology时候, Storm框架为我们提供了俩种模式:本地模式和集群模式

本地模式:(无需Storm集群,直接在java中即可运行,一般用于测试和开发阶段)执行运行main函数即可。

LocalCluster localCluster = new LocalCluster();
localCluster.submitTopology(topologyName, config, topologyBuilder.createTopology());
localCluster.killTopology(topologyName);
localCluster.shutdown();

集群模式:(需要 Storm集群,把实现的java程序打包,然后 Topology进行提交)需要把应用打成jar,使用storm命令把 Topology提交到集群中去。

StormSubmitter.submitTopology(topologyName, config, topologyBuilder.createTopology());

六 Storm 与其他数据处理框架的区别

6.1 Storm与Hadoop的区别

Storm 只是一个实时分布式计算框架，但可以读取 hdfs 上的文件进行批量计算，另外 Storm 可以运行在 Yarn 上。Hadoop 的 hdfs 是分布式文件存储系统，mapreduce + Yarn 批量离线计算。

Storm 的计算任务为 Topology，提交到 Storm 集群上运行，除非手动执行 kill 命令，否则会一直运行下去。MapReduce 数据处理完毕，就会停止。

原理角度来讲：

Hadoop M/R基于 HDFS，需要切分输入数据、产生中间数据文件、排序、数据压缩、多份复制等，效率较低。
Storm 基于 ZeroMQ 这个高性能的消息通讯库，不持久化数据。

6.2 Storm与SparkStreaming的区别

Storm是真正的实时处理框架，针对数据，来一条记录就处理一次，是一种流式数据处理技术框架，运行级别达到毫秒级别。

SparkStreaming是一种微批数据处理框架，运行级别达到秒级。

6.3 Storm与Flink的区别

比较项	Storm	Flink
状态管理	无状态，需用户自行进行状态管理	有状态
窗口支持	对事件窗口支持较弱，缓存整个窗口的所有数据，窗口结束时一起计算	窗口支持较为完善，自带一些窗口聚合方法，并且会自动管理窗口状态。
消息投递	At Most Once At Least Once	At Most Once At Least Once Exactly Once
容错方式	ACK机制：对每个消息进行全链路跟踪，失败或超时进行重发。	检查点机制：通过分布式一致性快照机制，对数据流和算子状态进行保存。在发生错误时，使系统能够进行回滚。
应用现状	在美团点评实时计算业务中已有较为成熟的运用，有管理平台、常用 API 和相应的文档，大量实时作业基于 Storm 构建。	在美团点评实时计算业务中已有一定应用，但是管理平台、API 及文档等仍需进一步完善。

单线程的吞吐量:

(图片参考来自美团技术团队）

上图中蓝色柱形为单线程 Storm 作业的吞吐，橙色柱形为单线程 Flink 作业的吞吐。
Identity 逻辑下，Storm 单线程吞吐为 8.7 万条/秒，Flink 单线程吞吐可达 35 万条/秒。
当 Kafka Data 的 Partition 数为 1 时，Flink 的吞吐约为 Storm 的 3.2 倍；当其 Partition 数为 8 时，Flink 的吞吐约为 Storm 的 4.6 倍。
由此可以看出，Flink 吞吐约为 Storm 的 3-5 倍。

flink 是一个设计良好的框架，它不但功能强大，而且性能出色。此外它还有一些比较好设计，比如优秀的内存管理和流控。但是，flink 目前成熟度较低，还存在着不少问题，比如 SQL 支持比较初级；无法像 storm 一样在不停止任务的情况下动态调整资源；不能像 spark 一样提供很好的 streaming 和 static data 的交互操作等。

参考文档:

[1] apache.storm官网: http://storm.apache.org/index.html
[2] GitHub地址: https://github.com/apache/storm
[3] 冰河 . 海量数据处理与大数据技术实战 [M] . 第1版 . 北京: 北京大学出版社 , 2020-09
[4] netcobol
. Storm 从入门到精通 . CSDN . https://blog.csdn.net/netcobol/category_9273336.html?spm=1001.2014.3001.5482 ， 2018-04-03
[5] fct2001140269 . Storm与Flink的比较 . CSDN . https://blog.csdn.net/fct2001140269/article/details/84479740 , 2018-11-25

ØMQ: ZeroMQ(也称为ØMQ, 0mq或zmq)看起来像一个嵌入式的网络库（an embeddable networking library） ↩︎