Strom介绍

Strom是什么

      Storm是由专业数据分析公司BackType开发的一个分布式计算框架，可以简单、高效、可靠地处理大量的数据流。Twitter在2011年7月收购该公司，并于2011年9月底正式将Storm项目开源。软件核心部分使用Clojure开发，外围部分使用Java开发。

      Storm可以方便地在一个计算机集群中编写与扩展复杂的实时计算，Storm用于实时处理，就好比  用于批处理。Storm保证每个消息都会得到处理，而且它很快——在一个小集群中，每秒可以处理数以百万计的消息。更棒的是你可以使用任意编程语言来做开发。

Strom特点

?可扩展性强

?Storm 的可伸缩性可以让Storm 每秒处理的消息量达到很高100 万。实现计算任务的扩展，只需要在集群中添加机器，然后提高计算任务的并行度设置

?消息可靠性

?所有消息都可保证至少处理一次。如果出错了，消息可能处理不只一次，不过你永远不会丢失消息

?高容错

?Storm集群会关注工作节点状态，如果宕机了必要的时候会重新分配任务

?语言无关性

?Storm 虽然是使用Clojure 语言开发实现。但是，Storm 的处理逻辑和消息处理组件都可以使用任何语言来进行定义，这就是说任何语言的开发者都可以使用Storm

?本地模式

?在处理过程中完全模拟Storm集群。让开发者快速进行开发和单元测试

Strom架构

?zookeeper

?Zookeeper 负责Nimbus 和Supervisor 之间的所有协调工作

?Nimbus 进程和Supervisor 都是快速失败和无状态的。所有的状态要么在Zookeeper 里面，要么在本地磁盘上

?可以用kill -9来“杀死”Nimbus 和Supervisor 进程，然后再重启它们，它们可以继续工作，就好像什么都没有发生过似的

?Nimbus

?（控制节点）

?Nimbus将提交的Topology进行分片，分成一个个的Task，并将Task和Supervisor相关的信息提交到 zookeeper集群上

?Nimbus 负责在集群里面分发执行代码，分配工作给工作节点，并且监控任务的执行状态

?Supervisor

?（工作节点）

?Supervisor会去zookeeper集群上认领自己的Task，通知自己的Worker进程进行Task的处理

?Supervisor 会监听分配给自己所在机器的工作，根据需要启动/关闭工作进程

?每一个工作进程执行一个Topology 的一个子集，一个运行的Topology 由运行在很多机器上的很多工作进程组成

Storm概念-topology

在 Storm 中，一个实时计算应用程序的逻辑被封装在一个称为Topology 的对象中，也称为计算拓扑

一个Topology 是由一些Spout（消息的发送者）和Bolt（消息的处理者）组成图状结构，而链接Spouts 和Bolts 的则是Stream Groupings。

Storm概念-spouts

Spouts 是Topology中消息流的生产者，一般从MQ、数据库、文件系统等其他数据源读取数据，然后向bolts发射消息。spouts可以分为可靠和非可靠的两种

spout实现可以通过继承BaseRichSpout类或者其他Spout类来完成，也可以通过实现IRichSpout接口来实现，提供了如下函数：
 void open(Map conf, TopologyContext context, SpoutOutputCollector collector); 
  //初始化工作，如连接MQ和数据库 

  void close(); 

 //关闭资源，如断开数据库连接
  void nextTuple(); 

 //storm向spout发出请求，让spout发出tuple到输出器（output collector）
  void ack(Object msgId); 

//成功处理tuple时回调的方法，将消息队列中的消息移除，防止消息重复
  void fail(Object msgId); 

//处理tuple失败时回调的方法，将消息放回消息队列中然后在稍后时间里重新处理

void declareOutputFields(OutputFieldsDeclarer declarer)

//声明发送数据格式

Storm概念-Bolts

在Bolt 中可以完成如过滤、分类、聚集、计算、查询数据库等操作。Bolts 不仅可以接收消息，也可以像Spout 一样发射多条消息流。

  Bolt类接收由Spout或者其他上游Bolt类发来的Tuple，对其进行处理。Bolt组件的实现可以通过继承BasicRichBolt类或者IRichBolt接口等来完成 

  提供了如下函数：
 void prepare(Map conf, TopologyContext context, OutputCollector collector) 
  //初始化工作，如连接MQ和数据库 

  void cleanup() 

 //关闭资源，如断开数据库连接
  void execute(Tuple input) 

 //storm向spout发出请求，让spout发出tuple到输出器（output collector）
void declareOutputFields(OutputFieldsDeclarer declarer)

//声明发送数据格式

Storm概念-streams

一个没有边界的、源源不断的、连续的Tuple序列就组成了Stream

Tuple 是一种Storm中使用的数据结构，可以看作是没有方法的Java 对象。

 tuple 的字段类型可以为integer、long、short、byte、string、double、float、boolean 和byte array 等基本类型

Strom概念- Stream Groupings

Stream Grouping 就是用来定义一个Stream 应该如何分配给Bolts 上面的多个Tasks

?Shuffle Grouping

?(随机分组)

?随机选择一个Task来发送

?Fields Grouping

?（按字段分组）

?根据Tuple中Fields来做一致性hash，相同hash值的Tuple被发送到相同的Task

?All Grouping

?（广播发送）

?将每一个Tuple发送到所有的Task

?Global Grouping

?（全局分组）

?所有的Tuple会被发送到某个Bolt中的id最小的那个Task。

?Non Grouping

?（不分组）

?这种分组和Shuffle Grouping 是一样的效果

?Direct Grouping

?（直接分组）

?由发送者指定由消息接收者的哪个Task 处理这个消息。只有被声明为Direct Stream 的消息流可以声明这种分组方法

Storm-概念并行度

storm rebalance mytopology -n 5 -e blue-spout=3 -e yellow-bolt=10

Storm-topology运行流程

Storm vs Hadoop