storm并行度

 

 

1.storm并行度
  storm的并行是由非常多的supervisor完成的。
  storm的supervisor运行的是topology中的spout/bolt task。
  task是storm中进行计算的最小的运行单位，表示的是spout、bolt的运行实例。
  程序执行的最大粒度的运行单位是进程。在supervisor中，运行task的进程称作worker。
  supervisor节点上可以运行非常多的worker。
  在worker中可以运行线程的，这些线程称作executor。在executor中，运行task。
 
  总结一下，supervisor(节点)>worker(进程)>executor(线程)>task(实例)

 

下面以一个节点为图例说明下 storm并行度：

 

 

 

 

 

 

2 如何具体设置并行度：

 

2.1）supervisor: 在哪个节点上启动supervisor 则就会产生  进入zk内查看storm结果集

2.2)  worker：storm.yaml中，如下指定了 worker进程的端口，以及当前机器下能运行的work数量

       每个端口用于对应进程对外通讯的。

       如下配置可以不配置，在storm的 storm-core-xx.jar中的default.xml中做了默认配置。

 

supervisor.slots.ports:    // 指定storm通讯端口
      - 6701
      - 6702
      - 6703
      - 6704

 

       worker进程数量也可以通过config.setNumWorkers(workers)设置。(优先级更高)

 

 2.3)  executor的设置：

builder.setSpout(id, spout, parallelism_hint)： parallelism_hint设置spout的数量

builder.setBolt(id, bolt, parallelism_hint)设置的。parallelism_hint设置bolt的数量

 

 2.4)  task的设置：

   task是通过 spout/boltDeclarer.setNumTasks(num)设置对应spout/bolt的task个数。

 

默认情况下，每个supervisor启动4个worker，每个worker启动1个executor，每个executor中会有1个task。
翻译成人类语言：  每台机器启动4个进程，每个进程里面跑一个线程，每个线程来具体执行一个 bolt/spout的实例

 

即： 可以通过 config设置进程个数  通过TopologyBuilder设置spout/bolt 各自的线程个数， 也能通过TopologyBuilder创建的对应的
Spout/BoltDecvlarer 来设置每个线程运行的具体task个数。

 

设置代码如下：

 

public class MyLocalStormTopology {

	/**
	 * 组装火车 轨道  并让火车在轨道上行驶
	 * @throws InterruptedException 
	 */
	public static void main(String[] args) throws Exception {

		// 祖品列车
		TopologyBuilder topologyBuilder = new TopologyBuilder();
		SpoutDeclarer spoutDeclarer = topologyBuilder.setSpout("1", new MySpout2()); // 定义1号车厢
		//spoutDeclarer.setNumTasks(val);// 设置task数量
		//topologyBuilder.setSpout(id, spout, parallelism_hint); // 动态设置 spout的executor数量
		BoltDeclarer boltDeclarer = topologyBuilder.setBolt("2", new MyBolt1()).shuffleGrouping("1");// 定义2号车厢 并和1号车厢连接起来   
		//boltDeclarer.setNumTasks(num); 动态设置 task数量
		//topologyBuilder.setBolt(id, bolt, parallelism_hint); 动态设置 bolt的executor数量

		// 造出轨道 在集群中运行
		StormSubmitter stormSubmitter = new StormSubmitter();// storm集群执行
		//HashMap conf = new HashMap();
		Config config = new Config();
		//config.setNumWorkers(workers); 动态设置每台机器运行work进程数
		stormSubmitter.submitTopology(MyLocalStormTopology.class.getSimpleName(), config, topologyBuilder.createTopology());
		
	}

 

 

  worker和外部通信：

 

 

 

  worker调优：

 Config类对象设置如下参数，来调整 worker进入数据，处理数据的容量大小

 

 

 

 

 

 

 

sdfsdfds