纵有疾风起
人生不言弃

Storm实践


1.Storm命令

  在Linux中观直接输入Storm,不带任何参数信息,可以查看Storm命令。

  参考这里

  1.  activate  

  激活指定的拓扑Spout。语法:storm activate topology-name

  2. classpath

  打印出Storm客户端运行命令时使用的类路径(classpath),语法:storm classpath

  3. deactivate

  禁用指定的拓扑Spout。语法:storm deactivate topology-name

  4. dev-zookeeper

  以dev.zookeeper.path配置的值作为本地目录,以storm.zookeeper.port配置的值作为端口,启动一个 新的ZooKeeper服务,仅用来开发/测试。语法:storm dev-zookeeper

  5. drpc

  启动一个DRPC守护进程。语法:storm drpc  该命令应该使用daemontools或者monit工具监控运行。

  6.help

  storm help <command>

  7. jar

  storm jar topology-jar-path class …
  把Storm的jar文件和“~/.storm”的配置放到类路径(classpath)中,以便当拓扑提交时,               StormSubmitter会上传topology-jar-path的jar文件。
 

常用命令:

提交Topologies到远程集群机器
  命令格式:storm jar 【jar路径】 【拓扑包名.拓扑类名】 【拓扑名称】

  样例:storm jar /storm-starter.jar storm.starter.WordCountTopology wordcountTop

  #提交storm-starter.jar到远程集群,并启动wordcountTop拓扑。
停止Topologies
  命令格式:storm kill 【拓扑名称】
  样例:storm kill wordcountTop
  #杀掉wordcountTop拓扑。
  执行kill命令时可以通过-w [等待秒数]指定拓扑停用以后的等待时间。
启动nimbus后台程序
  命令格式:storm nimbus
启动supervisor后台程序
  命令格式:storm supervisor
启动drpc服务
  命令格式:storm drpc
启动ui服务
  命令格式:storm ui
启动REPL
  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
  #根据指定参数打印远程配置的值。
执行Shell脚本

  命令格式:storm shell resourcesdir command args

打印CLASSPATH
  命令格式:storm classpath
平衡拓扑
  storm rebalance topology-name
  再平衡使你重分配集群任务。这是个很强大的命令。比如,你向一个运行中的集群增加了节点。再平衡命令将会停用拓扑,然后在相应超时时间之后重分配工人,并重启拓扑。
 

2.定义拓扑

Topology和Nimbus        
    Topology是storm的核心理念,将spout和bolt组织成一个topology,运行在storm集群里,完成实时分析和计算的任务。这里我主要想介绍下topology部署到storm集群的大概过程。提交一个topology任务到Storm集群是通过StormSubmitter.submitTopology方法提交:

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

    我们将topology打成jar包后,利用bin/storm这个python脚本,执行如下命令:

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

    将jar包提交给storm集群。storm脚本会启动JVM执行Topology的main方法,执行submitTopology的过程。而submitTopology会将jar文件上传到nimbus,上传是通过socket传输。

 
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],那么最终是这样分配

    {1: [host1:port1] 2 : [host2:port1]           3 : [host1:port1] 4 : [host2:port1]}

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

[host1:port1 host2:port1 host1:port2 host2:port2]

,然后分配任务为

{1: host1:port1 , 2 : host2:port2}

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

 
  
 

文章转载于:https://www.cnblogs.com/kxdblog/p/4810481.html

原著是一个有趣的人,若有侵权,请通知删除

未经允许不得转载:起风网 » Storm实践

分享到: 生成海报
avatar

评论 抢沙发

评论前必须登录!

立即登录   注册

切换注册

登录

忘记密码 ?

切换登录

注册

我们将发送一封验证邮件至你的邮箱, 请正确填写以完成账号注册和激活