Storm应用系列之——Topology部署


本系列属个人原创，转载请注明！

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

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

根据前文介绍，我们知道，storm的任务是包装在topology类中，由nimbus提交分配到整个cluster。

Topology有两种大类提交部署方式：

  • 提交到本地模式，一般用于调试。该模式下由于是起在同一个JVM进程下，所以不要让其负载太高。
  • 提交到集群模式。

提交到本地模式

这个非常的简单。 1. 编写代码 
public class LocalRunningTopology extends ExclaimBasicTopo {

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

		LocalRunningTopology topo = new LocalRunningTopology();
		Config conf = new Config();
		conf.setDebug(true);

		LocalCluster cluster = new LocalCluster();
		cluster.submitTopology("test", conf, topo.buildTopology());
		Utils.sleep(100000);
		cluster.killTopology("test");
		cluster.shutdown();
	}
}
2. 直接就可以在IDE里面运行，也可以提交到nimbus上，用nimbus进行本地运行：./storm jar storm-samples.jar com.edi.storm.topos.LocalRunningTopology

提交到集群

1. 编写代码 
public class ClusterRunningTopology extends ExclaimBasicTopo {

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

		String topoName = "test";
		
		ClusterRunningTopology topo = new ClusterRunningTopology();
		Config conf = new Config();
		conf.setDebug(true);

		conf.setNumWorkers(3);

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

2. 编译jar包 因为我是Maven项目，所以直接 生成jar
3. 上传至nimbus部署 ./storm jar storm-samples.jar com.edi.storm.topos.ClusterRunningTopology

实际开发时常用提交模式

实际开发时，我们往往是把本地和集群混绑在一起，用传入参数以示区别 
public static void main(String[] args) throws Exception {
		
		ExclaimBasicTopo topo = new ExclaimBasicTopo();
		Config conf = new Config();
		conf.setDebug(false);

		if (args != null && args.length > 0) {
			conf.setNumWorkers(3);

			StormSubmitter.submitTopology(args[0], conf, topo.buildTopology());
		} else {

			LocalCluster cluster = new LocalCluster();
			cluster.submitTopology("test", conf, topo.buildTopology());
			Utils.sleep(100000);
			cluster.killTopology("test");
			cluster.shutdown();
		}
	}

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

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

可以。


IDE直接提交至集群

修改上面提交集群的代码如下：

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

		String topoName = "test";
		ExclaimBasicTopo topo = new ExclaimBasicTopo();
		Config conf = new Config();
		conf.setDebug(false);

		File jarFile = EJob.createTempJar(RemoteRunningTopology.class.getClassLoader().getResource("").getPath());
        ClassLoader classLoader = EJob.getClassLoader();
        Thread.currentThread().setContextClassLoader(classLoader);
        
		//System.setProperty("storm.jar", Class.forName("com.edi.storm.topos.RemoteRunningTopology").getProtectionDomain().getCodeSource().getLocation().getPath());
        System.setProperty("storm.jar", jarFile.toString());
		conf.setNumWorkers(5);
		conf.setDebug(false);
		conf.put(Config.NIMBUS_HOST, "10.1.110.24");
		//conf.put(Config.NIMBUS_THRIFT_PORT, 8889);
		StormSubmitter.submitTopology(topoName, conf, topo.buildTopology());
	}

起作用的部分主要有三点：

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中参数


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

)下的inbox目录。

/{storm.local.dir}
  |
  | - /nimbus
        |
        | - /inbox
              |
              | - /stormjar-{uuid}.jar

不论上传的包名字是什么，最终会变成stormjar-{uuid}.jar。

a. 检查Storm状态是否active

b. 检查是否有同名topology在运行

c. 检查是否有同id的spout和bolt，以及其id是否合法。任何一个id都不能以"_"开头，这种命名方式是系统保留的。

/{storm.local.dir}
    |
    | - /nimbus
          |
          | - /inbox
          | - /stormdist
                |
                | - /{topology-id}
                        |
                        | - /stormjar.jar -- 包含这个topology所有代码的jar包(从nimbus/inbox挪过来)
                        |
                        | -/stormcode.ser -- 这个topology对象的序列化
                        | -/stormconf.ser -- 运行这个topology的配置

/

  |
  | - /workerbeats
         |
         | - {topology-id}
                |
                | - /{task-id}  -- task的心跳信息，包括心跳的时间，task运行时间以及一些统计信息

/{storm.zookeeper.root}
  |
  | - /assignments
        |
        | - /{topology-id}

6. 保存toplogy信息到zookeeper 

/{storm.zookeeper.root}
    |
    | - /storms
          |
          | - /{topology-id}

7. supervisor因为监听了zookeeper上的目录，所以当它发现有topology时，会先把所有的topology的信息如jar等下到本地，并删除不再运行的topology的本地信息

/{storm.local.dir}
    |
    | - /supervisor
          |
          | - stormdist
               |
               | - {topology-id}
                      |
                      | - stormcode.ser
                      | - stormconf.ser
                      | - stormjar.jar

8. supervisor根据分配的任务，去启动worker去处理assignment

9. worker启动后，会去zookeeper上找其对应的task。同时根据task的outbound信息建立对外的socket连接，将来发送tuple就是从这些socket连接发出去的。

到这里，一个topology就已经完全部署和运转起来了。


相关内容