Storm实践

　　命令格式：storm jar 【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】

　　#提交storm-starter.jar到远程集群，并启动wordcountTop拓扑。

　　命令格式：storm kill 【拓扑名称】

　　样例：storm kill wordcountTop

　　#杀掉wordcountTop拓扑。

　　执行kill命令时可以通过-w [等待秒数]指定拓扑停用以后的等待时间。

　　命令格式：storm nimbus

　　命令格式：storm supervisor

　　命令格式：storm drpc

　　命令格式：storm ui

　　REPL — read-evaluate-print-loop。

　　虽然clojure可以作为一种脚本语言内嵌在java里面，但是它的首选编程方式是使用REPL，这是一个简单的命令行接口，使用它你可以输入你的命令，执行，然后查看结果， 你可以以下面这个命令来启动REPL：

　　命令格式：storm repl

　　命令格式：storm localconfvalue 【配置参数关键字】

　　举例：storm localconfvalue storm.zookeeper.servers

　　#根据指定参数打印本地配置的值。

　　命令格式：storm remoteconfvalue 【配置参数关键字】

　　举例：storm remoteconfvalue storm.zookeeper.servers

　　#根据指定参数打印远程配置的值。

　　命令格式：storm classpath

　　storm rebalance topology-name

　　再平衡使你重分配集群任务。这是个很强大的命令。比如，你向一个运行中的集群增加了节点。再平衡命令将会停用拓扑，然后在相应超时时间之后重分配工人，并重启拓扑。

 

StormSubmitter.submitTopology(name, conf, builder.createTopology());

bin/storm jar xxxx.jar com.taobao.MyTopology args

 

topology任务的分配过程(zookeeper路径说明忽略root):
　　1.在zookeeper上创建/taskheartbeats/{storm id} 路径，用于任务的心跳检测。storm对zookeeper的一个重要应用就是利用zk的临时节点做存活检测。task将定时刷新节点的时间戳，然后nimbus会检测这个时间戳是否超过timeout设置。
　　2.从topology中获取bolts,spouts设置的并行数目以及全局配置的最大并行数，然后产生task id列表，如[1 2 3 4]
　　3.在zookeeper上创建/tasks/{strom id}/{task id}路径，并存储task信息
　　4.开始分配任务（内部称为assignment)， 具体步骤：
　　　　 (1)从zk上获得已有的assignment(新的toplogy当然没有了）
　　　　 (2)查找所有可用的slot，所谓slot就是可用的worker，在所有supervisor上配置的多个worker的端口。
 　　　　(3)将任务均匀地分配给可用的worker，这里有两种情况：
 　　　　(a)task数目比worker多，例如task是[1 2 3 4],可用的slot只有[host1:port1 host2:port1]，那么最终是这样分配

　　，可以看到任务平均地分配在两个worker上。
(b)如果task数目比worker少，例如task是[1 2]，而worker有[host1:port1 host1:port2 host2:port1 host2:port2]，那么首先会将woker排序，将不同host间隔排列，保证task不会全部分配到同一个worker上，也就是将worker排列成

，然后分配任务为

(4)记录启动时间
(5)判断现有的assignment是否跟重新分配的assignment相同，如果相同，不需要变更，否则更新assignment到zookeeper的/assignments/{storm id}上。
5.启动topology，所谓启动，只是将zookeeper上/storms/{storm id}对应的数据里的active设置为true。
6.nimbus会检查task的心跳，如果发现task心跳超过超时时间，那么会重新跳到第4步做re-assignment。