图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Storm › Storm应用系列之——Topology部署
分享

Storm应用系列之——Topology部署

hiqj 发表于 2014-12-28 22:10:03 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 4 17515
本帖最后由 nettman 于 2014-12-28 22:49 编辑
问题导读：


1、Topology有哪两种提交部署方式？
2.本地提交和集群如何提交job？
3.本地提交和集群提交代码有什么不同？




本系列源码地址： https://github.com/EdisonXu/storm-samples

1、Topology有两种大类提交部署方式：
  • 提交到本地模式，一般用于调试。该模式下由于是起在同一个JVM进程下，所以不要让其负载太高。
  • 提交到集群模式。

1）、提交到本地模式
这个非常的简单。
a. 编写代码
  1.     public class LocalRunningTopology extends ExclaimBasicTopo {  

  2.       

  3.         public static void main(String[] args) throws Exception {  

  4.       

  5.             LocalRunningTopology topo = new LocalRunningTopology();  

  6.             Config conf = new Config();  

  7.             conf.setDebug(true);  

  8.       

  9.             LocalCluster cluster = new LocalCluster();  

  10.             cluster.submitTopology("test", conf, topo.buildTopology());  

  11.             Utils.sleep(100000);  

  12.             cluster.killTopology("test");  

  13.             cluster.shutdown();  

  14.         }  

  15.     }  
复制代码

b. 直接就可以在IDE里面运行，也可以提交到nimbus上，用nimbus进行本地运行：
  1. ./storm jar storm-samples.jar com.edi.storm.topos.LocalRunningTopology
复制代码



2)、提交到集群
a. 编写代码
  1.     public class ClusterRunningTopology extends ExclaimBasicTopo {  

  2.       

  3.         public static void main(String[] args) throws Exception {  

  4.       

  5.             String topoName = "test";  

  6.               

  7.             ClusterRunningTopology topo = new ClusterRunningTopology();  

  8.             Config conf = new Config();  

  9.             conf.setDebug(true);  

  10.       

  11.             conf.setNumWorkers(3);  

  12.       

  13.             StormSubmitter.submitTopology(topoName, conf, topo.buildTopology());  

  14.         }  

  15.     }  

  16. 

复制代码
b. 编译jar包
因为我是Maven项目，所以直接 mvn clean install 生成jar
c. 上传至nimbus部署
./storm jar storm-samples.jar com.edi.storm.topos.ClusterRunningTopology

2、实际开发时常用提交模式
实际开发时，我们往往是把本地和集群混绑在一起，用传入参数以示区别
  1.     public static void main(String[] args) throws Exception {  

  2.               

  3.             ExclaimBasicTopo topo = new ExclaimBasicTopo();  

  4.             Config conf = new Config();  

  5.             conf.setDebug(false);  

  6.       

  7.             if (args != null && args.length > 0) {  

  8.                 conf.setNumWorkers(3);  

  9.       

  10.                 StormSubmitter.submitTopology(args[0], conf, topo.buildTopology());  

  11.             } else {  

  12.       

  13.                 LocalCluster cluster = new LocalCluster();  

  14.                 cluster.submitTopology("test", conf, topo.buildTopology());  

  15.                 Utils.sleep(100000);  

  16.                 cluster.killTopology("test");  

  17.                 cluster.shutdown();  

  18.             }  

  19.         }
复制代码

这样，本地就可以不传参直接运行，而需要部署到集群时，打完包传到nimbus上运行命令：
./storm jar storm-samples.jar com.edi.storm.topos.ClusterRunningTopology <TOPO_NAME>
填上一个集群唯一的<TOPO_NAME>即可。

有人又说了，这样还不是很方便，我能不能直接在IDE里面提交到storm集群？
可以。

3、IDE直接提交至集群
修改上面提交集群的代码如下：
  1.     public static void main(String[] args) throws Exception {  

  2.       

  3.             String topoName = "test";  

  4.             ExclaimBasicTopo topo = new ExclaimBasicTopo();  

  5.             Config conf = new Config();  

  6.             conf.setDebug(false);  

  7.       

  8.             File jarFile = EJob.createTempJar(RemoteRunningTopology.class.getClassLoader().getResource("").getPath());  

  9.             ClassLoader classLoader = EJob.getClassLoader();  

  10.             Thread.currentThread().setContextClassLoader(classLoader);  

  11.               

  12.             //System.setProperty("storm.jar", Class.forName("com.edi.storm.topos.RemoteRunningTopology").getProtectionDomain().getCodeSource().getLocation().getPath());  

  13.             System.setProperty("storm.jar", jarFile.toString());  

  14.             conf.setNumWorkers(5);  

  15.             conf.setDebug(false);  

  16.             conf.put(Config.NIMBUS_HOST, "10.1.110.24");  

  17.             //conf.put(Config.NIMBUS_THRIFT_PORT, 8889);  

  18.             StormSubmitter.submitTopology(topoName, conf, topo.buildTopology());  

  19.         }  
复制代码

起作用的部分主要有三点：
1）. 设置系统变量"storm.jar"。这个变量的值代表要部署的Topology的jar包地址。
这个地址必须是文件，所以，我们就可以写完代码后自己打个jar包放在某个固定位置，然后IDE直接运行该topology去集群提交部署。
当然，也可以在代码中打jar，所以我这里的代码中加入了一个打包的Utilities类，EJob。
2）. 设置参数Config.NIMBUS_HOST，其值为nimbus的hostname或ip地址。
3）. 设置参数Config.NIMBUS_THRIFT_PORT，其值为nimbus上Thrift接口的地址
也就是nimbus的conf/storm.yaml中参数nimbus.thrift.port的值，前提是你配了。如果没配，可以不设。

这样就可以直接在IDE里面运行提交上去了。

4、Topology提交原理
Topology提交后发生了什么呢？这个原理要放在这里讲了。因为这直接关系到对Strom运行概念的理解。
1）. Nimbus$Iface的beginFileUpload,uploadChunk以及finishFileUpload方法将运行的包上传至其数据目录(storm.yaml中storm.local.dir对应的目录)下的inbox目录。
  1.     /{storm.local.dir}  

  2.       |  

  3.       | - /nimbus  

  4.             |  

  5.             | - /inbox  

  6.                   |  

  7.                   | - /stormjar-{uuid}.jar  

  8. 

复制代码
不论上传的包名字是什么，最终会变成stormjar-{uuid}.jar。
2）. Nimbus$Iface的submitTopology方法会负责对这个topology进行处理，首先是对Storm本身及topology进行一些校验：
a. 检查Storm状态是否active
b. 检查是否有同名topology在运行
c. 检查是否有同id的spout和bolt，以及其id是否合法。任何一个id都不能以"_"开头，这种命名方式是系统保留的。
3）. 建立topology的本地目录
  1.     /{storm.local.dir}  

  2.         |  

  3.         | - /nimbus  

  4.               |  

  5.               | - /inbox  

  6.               | - /stormdist  

  7.                     |  

  8.                     | - /{topology-id}  

  9.                             |  

  10.                             | - /stormjar.jar -- 包含这个topology所有代码的jar包(从nimbus/inbox挪过来)  

  11.                             |  

  12.                             | -/stormcode.ser -- 这个topology对象的序列化  

  13.                             | -/stormconf.ser -- 运行这个topology的配置  
复制代码
4）. 建立该topology在zookeeper上的心跳目录
nimbus老兄是个有责任心的人，它虽然最终会把任务分成一个个task让supervisor去做，但是它时刻在关注着大家的情况，所以它要求每个task每隔一定时间就要给它打个招呼（心跳信息），让它知道事情还在正常发展。如果有task超时不打招呼，nimbus会人为这个task不行了，然后进行重新分配。zookeeper上的心跳目录：
  1.     /<span style="font-family: Consolas, 'Liberation Mono', Courier, monospace;">{storm.zookeeper.root}</span>  

  2.       

  3.       |  

  4.       | - /workerbeats  

  5.              |  

  6.              | - {topology-id}  

  7.                     |  

  8.                     | - /{task-id}  -- task的心跳信息，包括心跳的时间，task运行时间以及一些统计信息  
复制代码
5）. 计算topology的工作量
nimbus会根据topology中给的parallelism hint参数，来给spout/bolt设定task数目，并分配相应的task-id，然后把分配号的task信息写到zookeeper上去：
  1.     /{storm.zookeeper.root}  

  2.       |  

  3.       | - /assignments  

  4.             |  

  5.             | - /{topology-id}  
复制代码
6）. 保存toplogy信息到zookeeper
  1.     /{storm.zookeeper.root}  

  2.         |  

  3.         | - /storms  

  4.               |  

  5.               | - /{topology-id}  
复制代码
7）. supervisor因为监听了zookeeper上的目录，所以当它发现有topology时，会先把所有的topology的信息如jar等下到本地，并删除不再运行的topology的本地信息
  1.     /{storm.local.dir}  

  2.         |  

  3.         | - /supervisor  

  4.               |  

  5.               | - stormdist  

  6.                    |  

  7.                    | - {topology-id}  

  8.                           |  

  9.                           | - stormcode.ser  

  10.                           | - stormconf.ser  

  11.                           | - stormjar.jar  
复制代码
8）. supervisor根据分配的任务，去启动worker去处理assignment9）. worker启动后，会去zookeeper上找其对应的task。同时根据task的outbound信息建立对外的socket连接，将来发送tuple就是从这些socket连接发出去的。
到这里，一个topology就已经完全部署和运转起来了


原文地址：http://blog.csdn.net/xeseo/article/details/18219183




回复

使用道具 举报

自定义广告语

已有(4)人评论

跳转到指定楼层
正序浏览
355815741 发表于 2014-12-29 09:20:01
学习了，谢谢分享~
回复

使用道具 举报

shawl84 发表于 2014-12-31 20:32:29
回复

使用道具 举报

diandidemeng 发表于 2015-1-31 16:47:43
好资料，谢谢分享。
回复

使用道具 举报

ainubis 发表于 2015-3-29 18:06:19
好资料，谢谢分享。
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
hiqj

实习版主

关注

  • 22

    主题

  • 44

    帖子

  • 10

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000