storm拓扑的并行度(parallelism)概念
来自: http://my.oschina.net/u/2000675/blog/613142
1 storm并行的基本概念
一个运行中的拓扑是由什么构成的:工作进程(worker processes),执行器(executors)和任务(tasks)
在一个 Storm 集群中,Storm 主要通过以下三个部件来运行拓扑:
工作进程(worker processes)(进程数)
执行器(executors)(线程数)
任务(tasks)(实例组件数)
storm集群中的一个机器可以运行一个或者多个worker,对应于一个或者多个topologies.1个worker进程运行1个或多个excutor线程。每个worker从属于一个topology.executor是单线程。每1个executor运行着相同的组件(spout或bolt)的1个或多个task。1个task执行着实际的数据处理。
下面是他们之间相互关系的简单图示。
2 是否需要提高workers数目
(1) 最好 一台机器上的一个topology只使用一个worker,主要原因时减少了worker之间的数据传输
(2) 有更多的worker可能会有更好的表现,这取决于你的瓶颈所在。每个worker都有通过一个线程将元组转移到其他的worker,所以如果你的瓶颈在CPU和每个worker正在处理大量的元组,更多的worker可能会提高你的吞吐量。
所以基本上没有明确的答案,你应该根据你的环境和设计来尝试不同的配置。
3 executor的数目
executor是真正的并行度(事实上的并行度)。(task数目是想要设置的并行度)
executor初始数目=spout数目+bolt数目+acker数目 (这些加起来也就是task数目。)
spout数目,bolt数目,acker数目运行时是不会变化的,但是executor数目可以变化。
4 是否需要提高TASK数目
TASK的存在只是为了topology扩展的灵活性,与并行度无关。
1个task执行着实际的数据处理。
1个worker进程执行一个拓扑的子集。1个worker进程从属于1个特定的拓扑,并运行着这个拓扑的1个或多个组件(spout或bolt)的1个或多个executor。一个运行中的拓扑包括集群中的许多台机器上的许多个这样的进程。
1个executor是1个worker进程生成的1个线程。它可能运行着1个相同的组件(spout或bolt)的1个或多个task。
1 个task执行着实际的数据处理,你用代码实现的每一个spout或bolt就相当于分布于整个集群中的许多个task。在1个拓扑的生命周期中,1个组 件的task的数量总是一样的,但是1个组件的executor(线程)的数量可以随着时间而改变。这意味着下面的条件总是成立:thread的数量 <= task的数量。默认情况下,task的数量与executor的数量一样,例如,Storm会在每1个线程运行1个task。
配置拓扑的并发度
注意Storm的术语"并发度(parallelism)"是特别用来描述所谓的parallelism hint的,这代表1个组件的初始的executor(线程)的数量。在此文档中我们使用术语"并发度"的一般意义来描述你不但可以配置executor的数量,还可以配置worker进程的数量,还可以是1个拓扑的task的数量。在用到并发度的狭义的定义时我们会特别提出。
下面的小节给出了一些不同的配置选项,还有你如何在代码中设置它们。有多种方法可以进行设置,表格列举了其中几种。Storm目前有下列的配置优先级:defaults.yaml < storm.yaml < 特定拓扑的配置 < 内部特定组件的配置 < 外部特定组件的配置。
Worker 数量
说明:拓扑在集群中运行所需要的工作进程数
配置选项:TOPOLOGY_WORKERS
在代码中如何使用(示例):
Config#setNumWorkers
Executors(线程)数量
说明:每个组件需要的执行线程数
配置选项:(没有拓扑级的通用配置项)
在代码中如何使用(示例):
TopologyBuilder#setSpout()
注意:从 Storm 0.8 开始 parallelism_hint 参数代表 executor 的数量,而不是 task 的数量
Tasks 数量
说明:每个组件需要的执行任务数
配置选项:TOPOLOGY_TASKS
在代码中如何使用(示例):
ComponentConfigurationDeclarer#setNumTasks()
以下是配置上述参数的一个简单示例代码:
topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2) .setNumTasks(4) .shuffleGrouping("blue-spout);
在上面的代码中,我们为 GreenBolt 配置了 2 个初始执行线程(executor)和 4 个关联任务(task)。这样,每个执行线程中会运行 2 个任务。如果你在设置 bolt 的时候不指定 task 的数量,那么每个 executor 的 task 数会默认设置为 1。
拓扑示例
下图显示了一个与实际应用场景很接近的简单拓扑的结构。这个拓扑由三个组件构成:一个名为 BlueSpout 的 spout,和两个名为 GreenBolt 和 YellowBolt 的 bolt。这些组件之间的关系是:BlueSpout 将它的输出发送到 GreenBolt 中,然后GreenBolt 将消息继续发送到 YellowBolt 中。
图中是一个包含有两个 worker 进程的拓扑。其中,蓝色的 BlueSpout 有两个 executor,每个 executor 中有一个 task,并行度为 2;绿色的 GreenBolt 有两个 executor,每个 executor 有两个 task,并行度也为2;而黄色的YellowBolt 有 6 个 executor,每个 executor 中有一个 task,并行度为 6,因此,这个拓扑的总并行度就是 2 + 2 + 6 = 10。具体分配到每个 worker 就有 10 / 2 = 5 个 executor。
上图中,GreenBolt 配置了 task 数,而 BlueSpout 和 YellowBolt 仅仅配置了 executor 数。下面是相关代码:
Config conf = new Config(); conf.setNumWorkers(2); // use two worker processestopologyBuilder.setSpout("blue-spout", new BlueSpout(), 2); // set parallelism hint to 2topologyBuilder.setBolt("green-bolt", new GreenBolt(), 2) .setNumTasks(4) .shuffleGrouping("blue-spout"); topologyBuilder.setBolt("yellow-bolt", new YellowBolt(), 6) .shuffleGrouping("green-bolt");StormSubmitter.submitTopology( "mytopology", conf, topologyBuilder.createTopology() );
当然,Storm 还有一些其他的配置项可以控制拓扑的并行度,包括:
TOPOLOGY_MAX_TASK_PARALLELISM:该选项设置了一个组件最多能够分配的 executor 数(线程数上限),一般用于在本地模式运行拓扑时测试分配线程的数量限制。你可以通过 Config#setMaxTaskParallelism() 来配置该参数。
如何修改运行中的拓扑的并行度
Storm 的一个很有意思的特点是你可以随时增加或者减少 worker 或者 executor 的数量,而不需要重启集群或者拓扑。这个方法就叫做再平衡(rebalance)。
有两种方法可以对一个拓扑执行再平衡操作:
使用 Storm UI
使用以下所示的客户端(CLI)工具
下面是使用 CLI 工具的一个简单示例:
## 重新配置拓扑 "mytopology",使得该拓扑拥有 5 个 worker processes,## 另外,配置名为 "blue-spout" 的 spout 使用 3 个 executor,## 配置名为 "yellow-bolt" 的 bolt 使用 10 个 executor。$ storm rebalance mytopology -n 5 -e blue-spout=3 -e yellow-bolt=10