JSTORM使用笔记

安装部署

zeromq

简单快速的传输层框架，安装以下：

wget http://download.zeromq.org/zeromq-2.1.7.tar.gz
tar zxf zeromq-2.1.7.tar.gz
cd zeromq-2.1.7
./configure
make
sudo make install
sudo ldconfig 

jzmq

应该是zmq的java包吧，安装步骤以下：

git clone git://github.com/nathanmarz/jzmq.git
cd jzmq
./autogen.sh
./configure
make
make install

zookeeper

针对大型分布式系统提供配置维护、名字服务、分布式同步、组服务等，能够保证：

1. 顺序性：客户端的更新请求都会被顺序处理
2. 原子性：更新操做要不成功，要不失败
3. 一致性：客户端不论链接到那个服务端，展示给它的都是同一个视图
4. 可靠性：更新会被持久化
5. 实时性：对于每一个客户端他的系统视图都是最新的

在zookeeper中有几种角色：

1. Leader：发起投票和决议，更新系统状态
2. Follower：响应客户端请求，参与投票
3. Observer：不参与投票，只同步Leader状态
4. Client：发起请求

在启动以前须要在conf下编写zoo.cfg配置文件，里面的内容包括：

1. tickTime：心跳间隔
2. initLimit：Follower和Leader之间创建链接的最大心跳数
3. syncLimit：Follower和Leader之间通讯时限
4. dataDir：数据目录
5. dataLogDir：日志目录
6. minSessionTimeout：最小会话时间（默认tickTime * 2）
7. maxSessionTimeout：最大会话时间（默认tickTime * 20）
8. maxClientCnxns：客户端数量
9. clientPort：监听客户端链接的端口
10. server.N=YYYY:A:B：其中N为服务器编号，YYYY是服务器的IP地址，A是Leader和Follower通讯端口，B为选举端口

在单机的时候能够直接将zoo_sample.cfg修改成zoo.cfg，而后使用启动服务便可（若是报错没有目录，手动建立便可）：

sudo ./zkServer.sh start

如今用netstat -na（或者是./zkCli.sh 127.0.0.1:2181）就能看到在监听指定的端口，那么zookeeper如今起来了。 

参考：

1. http://blog.csdn.net/shenlan211314/article/details/6170717
2. http://blog.csdn.net/hi_kevin/article/details/7089358
3. 下载地址：http://apache.dataguru.cn/zookeeper/zookeeper-3.4.6

jstorm

该系统是阿里巴巴在对storm作了重写和优化，在storm里面能运行的在jstorm里面也能运行，该系统擅长执行实时计算，并且基本上都在内存中搞定。进入正题，jstorm中有以下几种角色：

1. spout：源头。
2. bolt：处理器。
3. topology：由处理器、源头组成的拓扑网络（每条边就是一个订阅关系）。
4. tuple：数据。
5. worker：执行进程。
6. task：执行线程。
7. nimbus：分发代码、任务，监控集群运行状态
8. supervisor：监听nimbus的指令，接收分发代码和任务并执行

jstorm是用zookeeper来管理的，下面来看conf/storm.yaml中的经常使用配置：

1. storm.zookeeper.servers：zookeeper集群地址。
2. storm.zookeeper.root：zookeeper中storm的根目录位置。
3. storm.local.dir：用来存放配置文件、JAR等。
4. storm.messaging.netty.transfer.async.batch：在使用Netty的时候，设置是否一个batch中会有多个消息。
5. java.library.path：本地库的加载地址，好比zeromq、jzmq等。
6. supervisor.slots.ports：supervisor节点上的worker使用的端口号列表。
7. supervisor.enable.cgroup：是否使用cgroups来作资源隔离。
8. topology.buffer.size.limited：是否限制内存，若是不限制将使用LinkedBlockingDeque。
9. topology.performance.metrics：是否开启监控。
10. topology.alimonitor.metrics.post：是否将监控数据发送给AliMonitor。
11. topology.enable.classloader：默认禁用了用户自定义的类加载器。
12. worker.memory.size：worker的内存大小。

在把配置搞正确以后，就能够用bin中的脚原本启动节点服务了：

sudo ./jstorm nimbus
sudo ./jstorm supervisor

参考：

1. https://github.com/alibaba/jstorm/wiki/%E5%A6%82%E4%BD%95%E5%AE%89%E8%A3%85
2. storm编程入门：http://ifeve.com/getting-started-with-storm-5/

jstorm的架构

结构和hadoop的很像，总体看来以下（Nimbus负责控制、提交任务，Supervisor负责执行任务）：

为了作实时计算你须要创建topology，由计算节点组成的图：

在JStorm上的topology的生命周期以下：

1. 上传代码并作校验（/nimbus/inbox）；
2. 创建本地目录（/stormdist/topology-id/）；
3. 创建zookeeper上的心跳目录；
4. 计算topology的工做量（parallelism hint），分配task-id并写入zookeeper；
5. 把task分配给supervisor执行；
6. 在supervisor中定时检查是否有新的task，下载新代码、删除老代码，剩下的工做交个小弟worker；
7. 在worker中把task拿到，看里面有哪些spout/Bolt，而后计算须要给哪些task发消息并创建链接；
8. 在nimbus将topology终止的时候会将zookeeper上的相关信息删除；

在集群运行的时候要明白Worker、Executor、Task的概念，固然消息被传递的时候其实发起者、接收者都是Task，而真正执行的是Executor（能够理解为一个线程），由它来轮询其中的Spout/Bolt：

在jstorm中经过ack机制来保证数据至少被处理一次，简单来讲下ack：

在消息发、收的过程当中会造成一棵树状的结构，在一个消息收的时候发一个验证消息，发的时候也发一个验证消息，那么整体上每一个消息出现两次。那么ack机制就是将每一个消息的随机生成的ID进行异或，若是在某一时刻结果为0，那就说明处理成功。

以下图所示： 

须要补充一下：虽然ack算是随机算法，可是出错的几率极低，可是系统应该具有在出错以后矫正的能力（甚至检查是否出错）。ack机制保证了消息会被处理，可是不能保证只处理一次&顺序处理，在须要的情形就有了事务的概念：

码代码

基本用法

所谓普通模式是指不去使用JStorm为开发人员提供的高级抽象，用其提供的原生的接口进行开发，主要涉及到的接口有：

1. ISpout：数据源头接口，jstorm会不断调用nextTuple方法来获取数据并发射出去。
   1. open：在worker中初始化该ISpout时调用，通常用来设置一些属性：好比从spring容器中获取对应的Bean。
   2. close：和open相对应（在要关闭的时候调用）。
   3. activate：从非活动状态变为活动状态时调用。
   4. deactivate：和activate相对应（从活动状态变为非活动状态时调用）。
   5. nextTuple：JStorm但愿在每次调用该方法的时候，它会经过collector.emit发射一个tuple。
   6. ack：jstorm发现msgId对应的tuple被成功地完整消费会调用该方法。
   7. fail：和ack相对应（jstorm发现某个tuple在某个环节失败了）。
2. IBolt：数据处理接口，jstorm将消息发给他并让其处理，完成以后可能整个处理流程就结束了，也可能传递给下一个节点继续执行。
   1. prepare：对应ISpout的open方法。
   2. cleanup：对应ISpout的close方法（吐槽一下，搞成同样的名字会死啊...）。
   3. execute：处理jstorm发送过来的tuple。
3. TopologyBuilder：每一个jstorm运行的任务都是一个拓扑接口，而builder的做用就是根据配置文件构建这个拓扑结构，更直白就是构建一个网。
   1. setSpout：添加源头节点并设置并行度。
   2. setBolt：添加处理节点并设置并行度。

由于还存在多种其余类型的拓扑结构，那么在builder这个环节固然不能乱传，在基本用法要去实现IRichSpout、IRichBolt接口，他们并无新增任何的方法，仅仅是用来区分类型。既然是拓扑结构那么应该是一个比较复杂的网络，其实这个是在builder中完成的，其中setSpout/setBolt返回的结果实际上是InputDeclarer对象，在其中定义了N个流分组的策略：

public T fieldsGrouping(String componentId, String streamId, Fields fields); // 字段分组
public T globalGrouping(String componentId, String streamId); // 全局分组
public T shuffleGrouping(String componentId, String streamId); // 随机分组
public T localOrShuffleGrouping(String componentId, String streamId); // 本地或随机分组
public T noneGrouping(String componentId, String streamId); // 无分组
public T allGrouping(String componentId, String streamId); // 广播分组
public T directGrouping(String componentId, String streamId); // 直接分组
// 自定义分组  
public T customGrouping(String componentId, CustomStreamGrouping grouping);  
public T customGrouping(String componentId, String streamId, CustomStreamGrouping grouping);  
public T grouping(GlobalStreamId id, Grouping grouping);  

经过这些接口，咱们能够一边增长处理节点、一边指定其消费哪些消息。

批量用法

基本的用法是每次处理一个tuple，可是这种效率比较低，不少状况下是能够批量获取消息而后一块儿处理，批量用法对这种方式提供了支持。打开代码能够很明显地发现jstorm和storm的有着不小的区别：

// storm 中的定义
public interface IBatchSpout extends Serializable {
	void open(Map conf, TopologyContext context);
	void emitBatch(long batchId, TridentCollector collector);// 批次发射tuple
	void ack(long batchId); // 成功处理批次
	void close();
	Map getComponentConfiguration();
	Fields getOutputFields();
}

// jstorm中的定义
public interface IBatchSpout extends IBasicBolt, ICommitter, Serializable {
}

另外若是用批次的话就须要改用BatchTopologyBuilder来构建拓扑结构，在IBatchSpout中主要实现的接口以下：

1. execute：虽然和IBolt中名字、参数一致，可是增长了一些默认逻辑
   1. 入参的input.getValue(0)表示批次（BatchId）。
   2. 发送消息时collector.emit(new Values(batchId, value))，发送的列表第一个字段表示批次（BatchId）。
2. commit：批次成功时调用，常见的是修改offset。
3. revert：批次失败时调用，能够在这里根据offset取出批次数据进行重试。

Transactional Topology

事务拓扑并非新的东西，只是在原始的ISpout、IBolt上作了一层封装。在事务拓扑中以并行（processing）和顺序（commiting）混合的方式来完成任务，使用Transactional Topology能够保证每一个消息只会成功处理一次。不过须要注意的是，在Spout须要保证可以根据BatchId进行屡次重试，在这里有一个基本的例子，这里有一个不错的讲解。

Trident

此次一种更高级的抽象（甚至不须要知道底层是怎么map-reduce的），所面向的再也不是spout和bolt，而是stream。主要涉及到下面几种接口：

1. 在本地完成的操做
   1. Function：自定义操做。
   2. Filters：自定义过滤。
   3. partitionAggregate：对同批次的数据进行local combiner操做。
   4. project：只保留stream中指定的field。
   5. stateQuery、partitionPersist：查询和持久化。
2. 决定Tuple如何分发到下一个处理环节
   1. shuffle：随机。
   2. broadcast：广播。
   3. partitionBy：以某一个特定的field进行hash，分到某一个分区，这样该field位置相同的都会放到同一个分区。
   4. global：全部tuple发到指定的分区。
   5. batchGlobal：同一批的tuple被放到相同的分区（不一样批次不一样分区）。
   6. partition：用户自定义的分区策略。
3. 不一样partition处理结果的汇聚操做
   1. aggregate：只针对同一批次的数据。
   2. persistentAggregate：针对全部批次进行汇聚，并将中间状态持久化。
4. 对stream中的tuple进行从新分组，后续的操做将会对每个分组独立进行（相似sql中的group by）
   1. groupBy
5. 将多个Stream融合成一个
   1. merge：多个流进行简单的合并。
   2. join：多个流按照某个KEY进行UNION操做（只能针对同一个批次的数据）。

在这里有一个jstorm中使用Trident的简单例子。

DRPC

 

 

 

 

 

 

问题排查

 

 

 

模型设计

在不少的实际问题中，咱们面对的模型都是大同小异，下面先来看问题是什么：

问题描述

一、在流式计算中常常须要对一批的数据进行汇总计算，若是用SQL来描述就是：

SELECT MIN(status) FROM my_table GROUP BY order_id

在用JStorm来实现这一条简单的SQL时，面对的是一条一条的数据库变化的消息（这里须要保证有序消费），其实至关于在一堆的消息上面作了一个嵌套的SQL查询，用一张图表示以下：

二、

 

　　

 

 

 

 

----- updating -----