Storm拓扑的并行度(parallelism)介绍
Storm分为3个主要实体,用于在Storm集群中运行拓扑
- 工作进程:Worker Process,也称为Worker
- 执行器:Executor,即线程Thread
- 任务:Task
工作进程、执行器、任务三者之间关系如下图:
1.topology配置介绍
topology由一个或者多个Spout/Bolt组件构成
运行中的topology由一个或者多个Supervisor节点中的Worker构成。默认情况下一个Supervisor节点运行4个Worker,由defaults.yaml/storm.yaml中的属性决定:
supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703
在代码中可以使用new Config().setNumberWorkers(3)进行设定,但是最大数量不能超过配置的supervisor.slots.ports数量。
Worker为特定拓扑的一个或者多个组件Spout/Bolt产生一个或者多个Executor。默认情况下一个Worker运行一个Executor。
Executor为特定拓扑的一个或者多个组件Spout/Bolt实例运行一个或者多个Task。默认情况下一个Executor运行一个Task。
Task执行真正的数据处理,代码中实现的每个Spout/bolt作为很多任务跨集群执行,一个Spout/Bolt组件的Task数量始终贯穿Topology的整个生命周期,但是一个Spout/Bolt组件的Executor数量会随着时间而改变。这意味着Thread<=Task条件成立。默认情况下Task数量与Executor数量相同,即Storm会使用每个Executor运行一个Task。
2.配置拓扑的并行度
- 工作进程Worker数量
Config config = new Config(); config.setNumWorkers(3); //注意此参数不能大于supervisor.slots.ports数量。
- 执行器Executor数量
TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(id, spout, parallelism_hint); //设置Spout的Executor数量参数parallelism_hint builder.setBolt(id, bolt, parallelism_hint); //设置Bolt的Executor数量参数parallelism_hint
- 任务Task数量
TopologyBuilder builder = new TopologyBuilder(); builder.setSpout(id, spout, parallelism_hint).setNumTasks(val); //设置Spout的Executor数量参数parallelism_hint,Task数量参数val builder.setBolt(id, bolt, parallelism_hint).setNumTasks(val); //设置Bolt的Executor数量参数parallelism_hint,Task数量参数val
3.改变运行中拓扑的并行度
Storm一个很好的特性是可以增加或减少工作进程Worker和Executor的数量而不需要重启集群或拓扑,这样的行为成为再平衡(rebalancing)。目前有两种方式可实现拓扑再平衡,如下: 使用Storm的WebUI 使用Storm的命令行工具,如下 # 重新配置拓扑 # “myTopology” 拓扑使用5个Worker进程 # “blue-spout” Spout使用3个Executor # “yellow-blot” Bolt使用10个Executor storm rebalance myTopology -n 5 -e blue-spout=3 -e yellow-blot=10