分布式实时计算系统Storm架构与运行原理

2018-06-23 来源：

一、Storm简介

Storm是一个免费并开源的分布式实时计算系统。利用Storm可以很容易做到可靠地处理无限的数据流，像Hadoop批量处理大数据一样，Storm可以实时处理数据。

Storm 很简单，可用于任意编程语言。Apache Storm 采用 Clojure 开发。Storm 有很多应用场景，包括实时数据分析、联机学习、持续计算、分布式 RPC、ETL 等。

Hadoop(大数据分析领域无可争辩的王者)专注于批处理?这种模型对许多情形(比如为网页建立索引)已经足够,但还存在其他一些使用模型,它们需要来自高度动态的来源的实时信息?为了解决这个问题,就得借助 Nathan Marz 推出的 storm(现在已经被Apache孵化)storm 不处理静态数据,但它处理连续的流数据。

storm特点：

编程简单：开发人员只需要关注应用逻辑，而且跟Hadoop类似，Storm提供的编程原语也很简单

高性能，低延迟：可以应用于广告搜索引擎这种要求对广告主的操作进行实时响应的场景。

分布式：可以轻松应对数据量大，单机搞不定的场景

可扩展：随着业务发展，数据量和计算量越来越大，系统可水平扩展

容错：单个节点挂了不影响应用

消息不丢失：保证消息处理

storm与hadoop的比较：

1.Storm用于实时计算，Hadoop用于离线计算。

2. Storm处理的数据保存在内存中，源源不断;Hadoop处理的数据保存在文件系统中，一批一批。

3. Storm的数据通过网络传输进来;Hadoop的数据保存在磁盘中。

4. Storm与Hadoop的编程模型相似

二、Storm集群架构

Storm集群采用主从架构方式，主节点是Nimbus，从节点是Supervisor，有关调度相关的信息存储到ZooKeeper集群中，架构如下图所示：

Nimbus

Storm集群的Master节点，负责分发用户代码，指派给具体的Supervisor节点上的Worker节点，去运行Topology对应的组件(Spout/Bolt)的Task。

Supervisor

Storm集群的从节点，负责管理运行在Supervisor节点上的每一个Worker进程的启动和终止。通过Storm的配置文件中的supervisor.slots.ports配置项，可以指定在一个Supervisor上最大允许多少个Slot，每个Slot通过端口号来唯一标识，一个端口号对应一个Worker进程(如果该Worker进程被启动)。

Worker

运行具体处理组件逻辑的进程。Worker运行的任务类型只有两种，一种是Spout任务，一种是Bolt任务。

Task

worker中每一个spout/bolt的线程称为一个task. 在storm0.8之后，task不再与物理线程对应，不同spout/bolt的task可能会共享一个物理线程，该线程称为executor。

ZooKeeper

用来协调Nimbus和Supervisor，如果Supervisor因故障出现问题而无法运行Topology，Nimbus会第一时间感知到，并重新分配Topology到其它可用的Supervisor上运行

三、Storm编程模型

Strom在运行中可分为spout与bolt两个组件，其中，数据源从spout开始，数据以tuple的方式发送到bolt，多个bolt可以串连起来，一个bolt也可以接入多个spot/bolt.运行时原理如下图：

Topology：Storm中运行的一个实时应用程序的名称。将 Spout、 Bolt整合起来的拓扑图。定义了 Spout和 Bolt的结合关系、并发数量、配置等等。

Spout：在一个topology中获取源数据流的组件。通常情况下spout会从外部数据源中读取数据，然后转换为topology内部的源数据。

Bolt：接受数据然后执行处理的组件,用户可以在其中执行自己想要的操作。

Tuple：一次消息传递的基本单元，理解为一组消息就是一个Tuple。

Stream：Tuple的集合。表示数据的流向。

四、Topology运行

在Storm中,一个实时应用的计算任务被打包作为Topology发布，这同Hadoop的MapReduce任务相似。但是有一点不同的是:在Hadoop中，MapReduce任务最终会执行完成后结束;而在Storm中，Topology任务一旦提交后永远不会结束，除非你显示去停止任务。计算任务Topology是由不同的Spouts和Bolts，通过数据流(Stream)连接起来的图?一个Storm在集群上运行一个Topology时，主要通过以下3个实体来完成Topology的执行工作：

(1). Worker(进程)

(2). Executor(线程)

(3). Task

下图简要描述了这3者之间的关系：

1个worker进程执行的是1个topology的子集(注：不会出现1个worker为多个topology服务)。1个worker进程会启动1个或多个executor线程来执行1个topology的component(spout或bolt)。因此，1个运行中的topology就是由集群中多台物理机上的多个worker进程组成的。

executor是1个被worker进程启动的单独线程。每个executor只会运行1个topology的1个component(spout或bolt)的task(注：task可以是1个或多个，storm默认是1个component只生成1个task，executor线程里会在每次循环里顺序调用所有task实例)。

task是最终运行spout或bolt中代码的单元(注：1个task即为spout或bolt的1个实例，executor线程在执行期间会调用该task的nextTuple或execute方法)。topology启动后，1个component(spout或bolt)的task数目是固定不变的，但该component使用的executor线程数可以动态调整(例如：1个executor线程可以执行该component的1个或多个task实例)。这意味着，对于1个component存在这样的条件：#threads<=#tasks(即：线程数小于等于task数目)。默认情况下task的数目等于executor线程数目，即1个executor线程只运行1个task。

总体的Topology处理流程图为：

下图是Storm的数据交互图，可以看出两个模块Nimbus和Supervisor之间没有直接交互。状态都是保存在Zookeeper上，Worker之间通过Netty传送数据。Storm与Zookeeper之间的交互过程，暂时不细说了。重要的一点:storm所有的元数据信息保存在Zookeeper中!

五、Storm Streaming Grouping

Storm中最重要的抽象，应该就是Stream grouping了，它能够控制Spot/Bolt对应的Task以什么样的方式来分发Tuple，将Tuple发射到目的Spot/Bolt对应的Task

目前，Storm Streaming Grouping支持如下几种类型：

Shuffle Grouping ：随机分组，尽量均匀分布到下游Bolt中

将流分组定义为混排。这种混排分组意味着来自Spout的输入将混排，或随机分发给此Bolt中的任务。shuffle grouping对各个task的tuple分配的比较均匀。

Fields Grouping ：按字段分组，按数据中field值进行分组;相同field值的Tuple被发送到相同的Task

这种grouping机制保证相同field值的tuple会去同一个task，这对于WordCount来说非常关键，如果同一个单词不去同一个task，那么统计出来的单词次数就不对了。“if the stream is grouped by the “user-id” field, tuples with the same “user-id” will alwaysGo to the same task”. —— 小示例

All grouping ：广播

广播发送，对于每一个tuple将会复制到每一个bolt中处理。

Global grouping ：全局分组，Tuple被分配到一个Bolt中的一个Task，实现事务性的Topology。

Stream中的所有的tuple都会发送给同一个bolt任务处理，所有的tuple将会发送给拥有最小task_id的bolt任务处理。

None grouping ：不分组

不关注并行处理负载均衡策略时使用该方式，目前等同于shuffle grouping,另外storm将会把bolt任务和他的上游提供数据的任务安排在同一个线程下。

Direct grouping ：直接分组指定分组

由tuple的发射单元直接决定tuple将发射给那个bolt，一般情况下是由接收tuple的bolt决定接收哪个bolt发射的Tuple。这是一种比较特别的分组方法，用这种分组意味着消息的发送者指定由消息接收者的哪个task处理这个消息。只有被声明为Direct Stream的消息流可以声明这种分组方法。而且这种消息tuple必须使用emitDirect方法来发射。消息处理者可以通过TopologyContext来获取处理它的消息的taskid (OutputCollector.emit方法也会返回taskid)。

另外，Storm还提供了用户自定义Streaming Grouping接口，如果上述Streaming Grouping都无法满足实际业务需求，也可以自己实现，只需要实现backtype.storm.grouping.CustomStreamGrouping接口，该接口重定义了如下方法：

List chooseTasks(int taskId, List values)

上面几种Streaming Group的内置实现中，最常用的应该是Shuffle Grouping、Fields Grouping、Direct Grouping这三种，使用其它的也能满足特定的应用需求。

六、可靠性

(1)、spout的可靠性

spout会记录它所发射出去的tuple，当下游任意一个bolt处理失败时spout能够重新发射该tuple。在spout的nextTuple()发送一个tuple时，为实现可靠消息处理需要给每个spout发出的tuple带上唯一ID，并将该ID作为参数传递给SpoutOutputCollector的emit()方法：collector.emit(new Values("value1","value2"), tupleID);实际上Values extends ArrayList<Object>

保障过程中，每个bolt每收到一个tuple，都要向上游应答或报错，在tuple树上的所有bolt都确认应答，spout才会隐式调用ack()方法表明这条消息（一条完整的流）已经处理完毕，将会对编号ID的消息应答确认；处理报错、超时则会调用fail()方法。

(2)、bolt的可靠性

bolt的可靠消息处理机制包含两个步骤：
a、当发射衍生的tuple，需要锚定读入的tuple
b、当处理消息时，需要应答或报错

可以通过OutputCollector中emit()的一个重载函数锚定或tuple：collector.emit(tuple, new Values(word)); 并且需要调用一次this.collector.ack(tuple)应答。

七、总结

最后再来梳理一下Storm中涉及的主要概念：

1.拓扑(Topology)：打包好的实时应用计算任务，同Hadoop的MapReduce任务相似。

2.元组(Tuple)：是Storm提供的一个轻量级的数据格式，可以用来包装你需要实际处理的数据。

3.流(Streams)：数据流(Stream)是Storm中对数据进行的抽象，它是时间上无界的tuple元组序列（无限的元组序列）?

4.Spout(喷嘴)：Storm中流的来源。Spout从外部数据源，如消息队列中读取元组数据并吐到拓扑里。

5.Bolts：在拓扑中所有的计算逻辑都是在Bolt中实现的。

6.任务(Tasks)：每个Spout和Bolt会以多个任务(Task)的形式在集群上运行。