任务调度接口定义：
| 1 | IScheduler{ |
| 2 | // conf为当前nimbus的stormp配置 |
| 3 | void prepare(Map conf); // 初始化 |
| 4 | // topologyies表示集群中所有topology信息,cluster表示当前集群包括用户自定义调度逻辑事所需的所有资源(Supervisor、Solt、以及任务分配情况) |
| 5 | void schedule(Topologies topologies,Cluster cluster); |
| 6 | }； |
EventScheduler
实现流程图:
功能：对资源进行均匀分配的调度器，实现了IScheduler接口, schedule方法实现如下
| 1 | defn– schedule[this ^Topologies topologyies ^Cluster cluster] |
| 2 | (schedule-topologies-evenly topologies cluster) |
schedule-topologies-evenly方法原型：
| 1 | defn schedule-topologies-evenly[^Topologies topologies ^Cluster cluster] |
方法说明：
- 调用cluster对象的needsSchedulingTopology方法获取需要进行任务调度的Topology集合，判读依据:Topology设置的NumWorkers数目是否大于已经分配给该Topology的Worker数目，以及该Topology尚未分配的Executor数目是否大于0.
- 对需要进行任务调度的Topology获取其topology-id，然后调用schedule-topology方法获取到new-assignment(<executor,node+port>集合)。
- 用node和port信息构造WorkerSlot对象并将作为slot.
- 对Executor集合中的每一项构造ExecutorDetail对象，并返回一个ExecutorDetails集合。
- 调用cluster的assign方法将计算出来的slot分配给与该Topology相对应的executors.
schedule-topology
方法原型：
| 1 | defn- schedule-topology [^TopologyDetails topology ~Cluster cluster] |
方法说明：
- 调用cluster的getAvailableSlots方法获取当前集群可用的slot资源（集群中还没使用的Supervisor端口），并转换为<node,port>集合(available-slots).
- 将topology中的ExecutorDetails集合转换为<start-task-id,end-task-id>集合。
- 调用get-alive-assigned-node+port->executors方法获取当前topology已经分配的资源情况，返回<node+port,executors>集合(alive-assigned)。
- 获取当前topology可以使用的slot数目，topology设置的worker数目与当前available-slots数目加上alive-assigned数据二者的最小值(total-slots-to-use)。
- 对available-slots进行排序，计算需要分配的solt数目（total-slots-to-use减去alive-assigned），从排序后的solt中顺序获取需要分配的solt做为reassign-solts.
- 比较all-executors跟已分配的Executor集合间的差异，获取需要进行分配的Executor集合，做为reassign-executors.
- 将计算出来的reassign-solts与reassign-executor进行关联，转换为<executor,slot>映射集合（映射方式为：使executor均匀的分布在slot上），保存到ressignment中.
时间： 2024-11-05 12:09:35