Storm：分布式流式计算框架

Storm是一个分布式的、高容错的实时计算系统。Storm适用的场景：

1. Storm可以用来用来处理源源不断的消息，并将处理之后的结果保存到持久化介质中。
2. 由于Storm的处理组件都是分布式的，而且处理延迟都极低，所以可以Storm可以做为一个通用的分布式RPC框架来使用。(实时计算？)

Storm集群架构

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

• Nimbus：Storm集群的Master节点，负责分发用户代码，指派给具体的Supervisor节点上的Worker节点，去运行Topology对应的组件（Spout/Bolt）的Task。
• Supervisor：Storm集群的从节点，负责管理运行在Supervisor节点上的每一个Worker进程的启动和终止。
• ZooKeeper：
  • 存储客户端提供的topology任务信息，nimbus负责将任务分配信息写入Zookeeper，supervisor从Zookeeper上读取任务分配信息
  • 存储supervisor和worker的心跳（包括它们的状态），使得nimbus可以监控整个集群的状态， 从而重启一些挂掉的worker
  • 存储整个集群的所有状态信息和配置信息。

组件抽象

我们先看一下，Topology提交到Storm集群后的运行时部署分布图，如下图所示：

通过上图我们可以看出，一个Topology的Spout/Bolt对应的多个Task可能分布在多个Supervisor的多个Worker内部。而每个Worker内部又存在多个Executor，根据实际对Topology的配置在运行时进行计算并分配。

• Topology：Storm对一个分布式计算应用程序的抽象，目的是通过一个实现Topology能够完整地完成一件事情（从业务角度来看）。一个Topology是由一组静态程序组件（Spout/Bolt）、组件关系Streaming Groups这两部分组成。
• Spout：描述了数据是如何从外部系统（或者组件内部直接产生）进入到Storm集群
• Bolt：描述了与业务相关的处理逻辑。
• Task：Spout/Bolt在运行时所表现出来的实体，都称为Task(多个)
• Worker：运行时Task所在的一级容器，Executor运行于Worker中，一个Worker对应于Supervisor上创建的一个JVM实例(和Spark一样的概念)
• Executor：运行时Task所在的直接容器，在Executor中执行Task的处理逻辑

数据流

storm是流式计算框架，数据源源不断地到来。storm中，每条消息称为元组，我们可以把这条消息灵活地看作KV结构。

容错

一般而言，数据在节点间的传递次数氛围以下三种，storm根据用户指定，选择这三种类型的数据保证：

1. 至少一次：节点收到相同的数据一次或者多次
2. 至多一次：节点最多收到一次数据(S4)，没有容错
3. 只有一次：计算正确性的必须保证(strom特有的可以选择数据只被计算一次，防止有些有状态的任务，多次计算后出错)

至少送达一次

storm在保证数据被传递到所有节点方面做得非常巧妙。

1. 对于每条消息i，赋予64bit长度的ID。同时，在一张表T(表是逻辑上的，通过一致性hash来找到对应的消息)上记录这条消息的初始ID di = ID。
2. 在每个后继节点上，每产生一条消息，就会生成一个随机ID。然后更新di = di ^ 消息输入ID ^ (消息输出ID1 ^ 消息输出ID2 ……)
3. 当di = 0，说明已经成功处理消息。
4. storm定期扫描T，对于没有成功更新为0的消息，重新发送。对于这种机制，我们可以发现，存在误判的概率2^(-64)

上面说得有点抽象，举例：

上面图12-10不对，系统表T是这样的

header 1	header 2
d1	[11011->0]
d2	[11101->0]

原理：每一条消息ID，我们都会异或两次，一次是前驱节点在产生这条消息的时候，一次是在后继节点消费这条消息后。

Trident

通过Trident API同时实现了状态持久化和“恰好送达一次”

1. 将多条数据封装为一个batch(这不成批处理了吗)，每个batch都有一个递增的ID(要有一个全局递增ID，需要zookeeper的帮助啊)。通过前面的至少发送一次机制，发现某个ID的batch处理失败，重新发送这个batch，ID不变
2. 各个计算节点在计算过程中通过ID提交状态，如果发现已经有这个ID的状态，则放弃本次计算。
3. ID必须顺序提交

storm高可用性(HA)

• 如果woker挂了，supervisor会重新创建
  • bolt任务失败直接重新启动
  • Spout任务失败，会再次从数据源拿数据
• 如果机器节点挂了，nimbus会把该节点上的task转移到其他节点
• 如果nimbus或者supervisor挂了，重启就行了。nimbus和supervisor被设计成无状态，- 状态都被存到zookeeper里面了
• 为防止nimbus挂掉，worker节点也挂掉，导致任务无法被nimbus转移到其他机器。nimbus也被设计成HA的，利用主从结构保证主节点挂了之后从节点一样能服务

Stream Groupings(shuffle)

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

1. Shuffle Grouping：随机分组，跨多个Bolt的Task，能够随机使得每个Bolt的Task接收到大致相同数目的Tuple，但是Tuple不重复
2. Fields Grouping：根据指定的Field进行分组 ，同一个Field的值一定会被发射到同一个Task上
3. Partial Key Grouping：与Fields grouping 类似，根据指定的Field的一部分进行分组分发
4. All Grouping：所有Bolt的Task都接收同一个Tuple（这里有复制的含义）
5. Global Grouping：所有的流都指向一个Bolt的同一个Task（也就是Task ID最小的）
6. None Grouping：不需要关心Stream如何分组，等价于Shuffle grouping
7. Direct Grouping：由Tupe的生产者来决定发送给下游的哪一个Bolt的Task ，这个要在实际开发编写Bolt代码的逻辑中进行精确控制
8. Local or Shuffle Grouping：如果目标Bolt有1个或多个Task都在同一个Worker进程对应的JVM实例中，则Tuple只发送给这些Task

最常用的应该是Shuffle Grouping(随机)、Fields Grouping(哈希)、Direct Grouping(用户指定)这三种

Topology并行度计算

conf.setNumWorkers(2); // 该Topology运行在Supervisor节点的2个Worker进程中

topologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); // 设置并行度为2，则Task个数为2*1

topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2)
               .setNumTasks(4)
               .shuffleGrouping("blue-spout"); // 设置并行度为2，设置Task个数为4 ，则Task个数为4

topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6)
               .shuffleGrouping("green-bolt"); // 设置并行度为6，则Task个数为6*1

一共有12个任务，10的并行度，2个work。所以一个work里面有5个executor，6个task(分别是1蓝，2绿，3黄)，storm会把同类型的Task尽量放到同一个Executor中运行，Task个数最少的开始分配。

work和excutor内部构造

• 同一work不同executor的两个task消息传递
  
• 不同work不同executor的两个task消息传递
  

上图我们还能看出同一executor不同task之间的消息传递

总结：

1. 同一work之内的消息传递都要通过executor的消息收发线程
2. 不同work的消息传递要通过work的收发线程
3. 每个Executor应该维护Task与所在的Executor之间的关系，这样才能正确地将Tuple传输到目的Bolt Task进行处理。