storm从入门到放弃(二),任务分配过程-核心机密

背景:目前就职于国内最大的IT咨询公司,恰巧又是毕业季,所在部门招了100多个应届毕业生,本人要跟部门新人进行为期一个月的大数据入职培训,特此将整理的文档分享出来。

原文和作者一起讨论:http://www.cnblogs.com/intsmaze/p/7275406.html

微信:intsmaze

storm从入门到放弃(二),任务分配过程-核心机密

集群环境

storm机器有4台节点(物理机),三台是supervisor,每一台supervisor上面启动4个work进程(JVM进程),一共有12个work进程。

Topology程序

public class WordCountTopologyMain {
    public static void main(String[] args) throws Exception {
        TopologyBuilder intsmaze= new TopologyBuilder();
        intsmaze.setSpout("spout", new RandomSentenceSpout(),3);
        intsmaze.setBolt("split", new SplitSentenceBolt(),9).shuffleGrouping("spout");
        intsmaze.setBolt("count", new WordCountBolt(),3).fieldsGrouping("split",new Fields("word"));

        Config conf = new Config();
        conf.setDebug(false);
        //定义你希望集群分配多少个工作进程给你来执行这个topology,这里3个进程(work)来运行15个execute(线程)
          conf.setNumWorkers(3);
        StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
    }
}
  程序中指定需要三个work来执行一共15个任务。客户端提交任务给nimbus后,nimbus从12个空闲的worker中挑选3个worker去运行15个任务。
 
  疑问:nimbus怎么知道集群中的supervisor中有多少个work,以及空闲的work?
  supervisor中的work数量在配置文件中定义好了,supervisor启动后会把本机的work信息存放到zookeeper目录下,这样nimbus就可以到zookeeper目录下获取work信息,也就知道哪些work可以使用了。
 
  哪些task运行在哪些worker上?task的命名和分配到work的算法代码在哪里?
  对这3组不同的task(一共15个task),进行编号,按照一个全局顺序编号,对编号根据设置的work数量进行取余来选择放到哪一个work上。
supervisor1上面的task编号为1(spout); ,,(bolt task split); (bolt task wordcount)
supervisor2上面的task编号为2(spout); ,,(bolt task split); (bolt task wordcount)
supervisor3上面的task编号为3(spout); ,,(bolt task split); (bolt task wordcount)
  这是一种最优解,这样可以保证每一个 spout task任务发送数据给下一个bolt任务能最大限度在同一个进程间传递信息,避免了向另一台节点的bolt发送消息带来的网络开销。所以为了保证能最大程度在本地local传递数据,设置的task的总数最好能整除worker数,每一个组件(spout/bolt)的并行度最好能整除worker数。
storm从入门到放弃(二),任务分配过程-核心机密
  一个worker只属于一个拓扑程序,这是因为,这个worker启动了,它就不是空闲的了,nimbus再分配新的拓扑程序时,只会从空闲的worker中选择。

Topology程序

public class WordCountTopologyMain {
    public static void main(String[] args) throws Exception {
        TopologyBuilder intsmaze= new TopologyBuilder();

        intsmaze.setSpout("spout", new RandomSentenceSpout(),3).setNumTasks(9);
        //3是说明该spout启动几个线程来运行。该组件每个线程执行3个task.

        intsmaze.setBolt("split", new SplitSentenceBolt(),9).shuffleGrouping("spout");
        //不指定默认一个线程一个task任务
         intsmaze.setBolt("count", new WordCountBolt(),3).fieldsGrouping("split",new Fields("word"));

        Config conf = new Config();
        conf.setDebug(false);
        //定义你希望集群分配多少个工作进程给你来执行这个topology,这里3个进程(work)来运行15个execute(线程)
          conf.setNumWorkers(3);
        StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
    }
}
  程序中指定需要三个work十五个线程来执行二十一个任务。客户端提交任务给nimbus后,nimbus从12个空闲的worker中挑选3个worker去调用十五个线程来执行二十一个任务。
  这个时候怎么分配,因为在spout中会有3个task在一个线程中。所以在spout的task中三个task算一组进程work选择。
  哪些task运行在哪些worker上?
  对这3组不同的task(一共21个task),进行编号,按照一个全局顺序编号,对编号根据设置的work数量进行取余来选择放到哪一个work上。
supervisor1上面的task编号为spout -,spout1-(spout); split -, split -, split -(bolt task split); wordcount -(bolt task wordcount)
supervisor2上面的task编号为spout2-,spout -(spout); split -, split -, split -(bolt task split); wordcount -(bolt task wordcount)
supervisor3上面的task编号为spout -,spout -(spout); split -, split -, split -(bolt task split); wordcount -(bolt task wordcount)

配置并行度

对于并发度的配置, 在storm里面可以在多个地方进行配置, 优先级为:
efaults.yaml < storm.yaml < topology-specific configuration < internal component-specific configuration < external component-specific configuration 
worker processes的数目, 可以通过配置文件和代码中配置, worker就是执行进程, 所以考虑并发的效果, 数目至少应该大于machines的数目。
executor的数目, component的并发线程数,只能在代码中配置(通过setBolt和setSpout的参数), 例如, setBolt("green-bolt", new GreenBolt(), 2) 。
tasks的数目, 可以不配置, 默认和executor1:1, 也可以通过setNumTasks()配置。
Topology的worker通过config设置,即执行该topology的worker(java)进程数。它可以通过storm rebalance 命令任意调整。
 storm从入门到放弃(二),任务分配过程-核心机密
3个组件的并发度加起来是10,就是说拓扑一共有10个executor,一共有2个worker,每个worker产生10 / 2 = 5条线程。绿色的bolt配置成2个executor和4个task。为此每个executor为这个bolt运行2个task。
上一篇:csu 10月 月赛 B 题 Scoop water


下一篇:entity framework 新手入门篇(1)-建立模型