Storm默认的任务调度器。实现如下:
1 | (defn –prepare [this conf]) |
2 | (defn –schedule [this ^Topologies topologies ^Cluster cluster] |
3 | (default-schedule topologies cluster)) |
default-schedule
方法原型:
1 | (defn default-schedule [^Topologies topologies ^Cluster cluster]) |
方法说明:
- 调用cluster对象的needsSchedulingTopology方法获取需要进行任务调度的Topology集合.
- 调用cluster的getAvailableSlots方法获取当前集群可用的slot资源(集群中还没使用的Supervisor端口),并转换为<node,port>集合(available-slots).
- 将topology中的ExecutorDetails集合转换为<start-task-id,end-task-id>集合存入all-executors.
- 调用get-alive-assigned-node+port->executors方法获取当前topology已经分配的资源情况,返回<node+port,executors>集合(alive-assigned).
- 调用slots-can-ressign方法对alive-assigned的slot信息进行判断,选择其中可被重新分配的slot集合并保存到can-ressign-slots变量中。
- 计算当前Topology所能使用的全部slot的数目,topology设置的worker数目与当前available-slots数目加上can-ressign-slots数据二者的最小值(total-slots-to-use)。
- 判断total-slots-to-use的数目是否大于当前已分配的slot数目(alive-assigned),若大于则调用bad-slots方法计算所有可能被释放的slot.
- 调用cluster的freeSlots方法释放前面计算出来的bad-slots。
- 调用EventScheduler的schedule-topologies-evenly方法将系统中的资源均匀分配该Topology.
调度流程图:
slots-can-reassign
功能:从已经分配给当前Topology的资源中过滤出可以继续使用的资源。
函数原型:
1 | (defn slots-can-reassign [^Cluster cluster slots]) |
方法说明:
- 参数slots为已分配的slots资源,<node,port>集合。
- 对传入的slots进行过滤,过滤方式:选判断slots的node信息是否存在于集群的黑名单中,如果不在,继续判断slot的port是否在与node相对应的Supervisor的所有可用端口列表中,如果在该slots就可以继续使用。
bad-slots
功能:计算一个Topology已经分配的资源中哪些是不再需要的。
函数原型:
1 | (defn- bad-slots [existing-slots num-executors num-workers]) |
参数说明:
Existing-slots:已分配给Topology的资源,<[node,port],executors>集合。
Num-executors:Topology的所有Executor(包括已分配和未分配的)。
Num-workers:Topology可使用的全部slot数目。
方法说明:
- 判断num-workers是否为0,如果是表示当前没有可供该Topology使用的slots并返回空集合。
- 定义distribution、keepers集合,调用integer-divided方法将num-executors均匀地分配到num-worksers中结果集保存到distribution集合中,结果集格式<executor-count,worker-count>, executor-count表示单个worker被分配Executor的个数,worker-count表示有多少个这样的worker,keepers集合默认为空.
- 对existing-slots中的每一项计算executor-count,然后根据distribution集合,及该executor-count为键获取值,若所获取值大于0,意味着存在这样的分配,这时将该<[node,port],executors>信息放入keepers中,同时将distribution中该executor-count的对应值减一。
- 从existing-slots中移除keepers中记录的需要继续维持的分配情况,若移除完之后还存在slot信息,则表明这些slot可以被释放,并将其转换为workerslot对象集合返回。
时间: 2024-10-09 23:49:00