JStorm介绍

一、简介

二、数据模型

JStorm通过一系列基本元素实现实时计算的目标，其中包括了Topology、Stream、Spout、Bolt等等。JStorm在模型上和MapReduce有很多相似的地方，下表从不同维度对JStorm和MapReduce进行了比较。

实时计算任务需要打包成Topology提交，和MapReduce Job相似，不同的是，MapReduce Job在计算完成后结束，而JStorm的Topology任务一旦提交永远不会结束，除非显式停止。

计算任务Topology是由不同的Spout和Bolt通过Stream连接起来的DAG图。下面是一个典型Topology的结构示意图：

其中：

Spout：JStorm的消息源。用于生产消息，一般是从外部数据源（如MQ/RDBMS/NoSQL/RTLog等）不间断读取数据并向下游发送消息。

Bolt：JStorm的消息处理者。用于为Topology进行消息处理，Bolt可以执行查询、过滤、聚合及各种复杂运算操作，Bolt的消息处理结果可以作为下游Bolt的输入不断迭代。

Stream：JStorm中对数据进行的抽象，它是时间上无界的Tuple元组序列。在Topology中Spout是Stream的源头，负责从特定数据源发射Stream；Bolt可以接收任意多个Stream输入然后进行数据的加工处理，如果需要Bolt还可以发射出新Stream给下游Bolt。

Tuple：JStorm使用Tuple作为数据模型，存在于任意两个有数据交互的组件（Spout/Bolt）之间。每个Tuple是一组具有各自名称的值，值可以是任何类型，JStorm支持所有的基本类型、字符串以及字节数组，也可以使用自定义类型（需实现对应序列化器）作为值类型。简单来说，Tuple就是一组实现了序列化器带有名称的Java对象集合。

Topology中每一个计算组件（Spout和Bolt）都有一个并行度，在创建Topology时指定（默认为1），JStorm在集群内分配对应个数的线程Task并行。

如上图示，既然对于Spout/Bolt都会有多个线程来并行执行，那么如何在两个组件（Spout和Bolt）之间发送Tuple会成为新的问题。

JStorm通过定义Topology时为每个Bolt指定输入Stream以及指定提供的若干种数据流分发（Stream Grouping）策略用来解决这一问题。

三、系统架构

JStorm与Hadoop相似，保持了Master/Slave的简洁优雅架构。与Hadoop不同，JStorm的M/S之间不是直接通过RPC交换心跳信息，而是借助ZK来实现，这样的设计虽然引入了第三方依赖，但是简化了Nimbus/Supervisor的设计，同时也极大提高了系统的容错能力。

整个JStorm系统中共存三类不同的Daemon进程，分别是Nimbus，Supervisor和Worker。

Nimbus：JStorm中的主控节点，Nimbus类似于MR的JT，负责接收和验证客户端提交的Topology，分配任务，向ZK写入任务相关的元信息，此外，Nimbus还负责通过ZK来监控节点和任务健康情况，当有Supervisor节点变化或者Worker进程出现问题时及时进行任务重新分配。Nimbus分配任务的结果不是直接下发给Supervisor，也是通过ZK维护分配数据进行过渡。特别地，JStorm 0.9.0领先Apache Storm实现了Nimbus HA，由于Nimbus是Stateless节点，所有的状态信息都交由ZK托管，所以HA相对比较简单，热备Nimbus subscribe ZK关于Master活跃状态数据，一旦发现Master出现问题即从ZK里恢复数据后可以立即接管。

Supervisor：JStorm中的工作节点，Supervisor类似于MR的TT，subscribe ZK分配到该节点的任务数据，根据Nimbus的任务分配情况启动/停止工作进程Worker。Supervisor需要定期向ZK写入活跃端口信息以便Nimbus及时监控。Supervisor不执行具体的数据处理工作，所有的数据处理工作都交给Worker完成。

Worker：JStorm中任务执行者，Worker类似于MR的Task，所有实际的数据处理工作最后都在Worker内执行完成。Worker需要定期向Supervsior汇报心跳，由于在同一节点，同时为保持节点的无状态，Worker定期将状态信息写入本地磁盘，Supervisor通过读本地磁盘状态信息完成心跳交互过程。Worker绑定一个独立端口，Worker内所有单元共享Worker的通信能力。

Nimbus、Supervisor和Worker均为Stateless节点，支持Fail-Fast，这为JStorm的扩展性和容错能力提供了很好的保障。

还剩一个问题是Topology的各个计算组件（Spout/Bolt）如何映射到计算资源上。梳理这个问题前需要先明确Worker/Executor/Task之间的关系：

0、Worker：完整的Topology任务是由分布在多个Supervisor节点上的Worker进程（JVM）来执行，每个Worker都执行且仅执行Topology任务的一个子集。

1、Executor：Worker内部会有一个或多个Executor，每个Executor对应一个线程。Executor包括SpoutExecutor和BoltExecutor，同一个Worker里所有的*Executor只能属于某一个Topology里的执行单元。

2、Task：执行具体数据处理实体，也就是用户实现的Spout/Blot实例。一个Executor可以对应多个Task，定义Topology时指定，默认Executor和Task一一对应。这就是说，系统中Executor数量一定是小于等于Task数量（#Executor≤#Task）。

下图给出了一个简单的例子，上半部分描述的是Topology结构及相关说明，其中定义了整个Topology的worker=2，DAG关系，各个计算组件的并行度；下半部分描述了Topology的Task在Supervisor节点的分布情况。从中可以看出Topology到Executor之间的关系。

0、Worker数在提交Topology时在配置文件中指定；

例：#Worker=2

1、执行线程/Executor数在定义Topology的各计算组件并行度时决定，可以不指定，默认为1。其中各个计算组件的并行度之和即为该Topology执行线程总数。

例：#Executor=sum(#parallelism hint)=2+2+6=10

2、Task数目也在定义Toplogy时确定，若不指定默认每个Executor线程对应一个Task，若指定Task数目会在指定数目的线程里平均分配。

例：#Task=sum(#task)=2+4+6=12，其中Executor4={Task0,Task1}

四、 关键流程

0、Topology提交

JStorm为用户提供了StormSubmitter. submitTopology用来向集群提交Topology，整个提交流程：

1、任务调度策略

2、Acker机制

为保证无数据丢失，Storm/JStorm使用了非常漂亮的可靠性处理机制，如图当定义Topology时指定Acker，JStorm除了Topology本身任务外，还会启动一组称为Acker的特殊任务，负责跟踪Topolgogy DAG中的每个消息。每当发现一个DAG被成功处理完成，Acker就向创建根消息的Spout任务发送一个Ack信号。Topology中Acker任务的并行度默认parallelism hint=1，当系统中有大量的消息时，应该适当提高Acker任务的并行度。

Acker按照Tuple Tree的方式跟踪消息。当Spout发送一个消息的时候，它就通知对应的Acker一个新的根消息产生了，这时Acker就会创建一个新的Tuple Tree。当Acker发现这棵树被完全处理之后，他就会通知对应的Spout任务。

Acker任务保存了数据结构Map

>，

其中MessageID是Spout根消息ID，TaskID是Spout任务ID，Value表示一个64bit的长整型数字，是树中所有消息的随机ID的异或结果。通过TaskID，Acker知道当消息树处理完成后通知哪个Spout任务，通过MessageID，Acker知道属于Spout任务的哪个消息被成功处理完成。Value表示了整棵树的的状态，无论这棵树多大，只需要这个固定大小的数字就可以跟踪整棵树。当消息被创建和被应答的时候都会有相同的MessageID发送过来做异或。当Acker发现一棵树的Value值为0的时候，表明这棵树已经被成功处理完成。

需要指出的是，Acker并不是必须的，当实际业务可以容忍数据丢失情况下可以不用Acker，对数据丢失零容忍的业务必须打开Acker，另外当系统的消息规模较大是可适当增加Acker的并行度。

3、故障恢复

0）节点故障

1）任务失败

五、基础接口

这里把几个基础接口中注释摘出来说明其的作用：

针对前面例子中的Topology这里给出一个简单的实现，其中略去了BlueSpout/GreeBolt/YellowBolt的具体实现，更多参考这里。

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 

public static void main (String[] args){ Config conf = new Config(); // use two worker processes conf.setNumWorkers(2); // set parallelism hint to 2 topologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2) .setNumTasks(4) .shuffleGrouping("blue-spout"); topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6) .shuffleGrouping("green-bolt"); StormSubmitter.submitTopology( "mytopology", conf, topologyBuilder.createTopology()); }

JStorm更多包括事务在内的接口详见源码。

六、结语

本文对JStorm做了简单介绍，有错误之处敬请指正。