Storm-源码分析- Scheduler (backtype.storm.scheduler)

首先看看IScheduler接口的定义, 主要实现两个接口, prepare和schedule

对于schedule的参数注释写的非常清楚,
topologies包含所有topology的静态信息, 而cluster中包含了topology的运行态信息
根据他们就可以来判断如何assignment

package backtype.storm.scheduler;
import java.util.Map;
public interface IScheduler {
void prepare(Map conf);

/**
* Set assignments for the topologies which needs scheduling. The new assignments is available
* through <code>cluster.getAssignments()</code>
*
*@param topologies, all the topologies in the cluster, some of them need schedule. Topologies object here
*       only contain static information about topologies. Information like assignments, slots are all in
*       the <code>cluster</code>object.
*@param cluster, the cluster these topologies are running in. <code>cluster</code> contains everything user
*       need to develop a new scheduling logic. e.g. supervisors information, available slots, current
*       assignments for all the topologies etc. User can set the new assignment for topologies using
*       <code>cluster.setAssignmentById</code>
*/
void schedule(Topologies topologies, Cluster cluster);
}

DefaultScheduler

DefaultScheduler, 实现backtype.storm.scheduler.IScheduler接口

(ns backtype.storm.scheduler.DefaultScheduler
(:gen-class
:implements [backtype.storm.scheduler.IScheduler]))
(defn -prepare [this conf]
)
(defn -schedule [this ^Topologies topologies ^Cluster cluster]
(default-schedule topologies cluster))

 

下面看看default-schedule做了些什么?

(defn default-schedule [^Topologies topologies ^Cluster cluster]
(let [needs-scheduling-topologies (.needsSchedulingTopologies cluster topologies)] ;;1,取出需要scheduling的topologies
(doseq [^TopologyDetails topology needs-scheduling-topologies
:let [topology-id (.getId topology)
available-slots (->> (.getAvailableSlots cluster)
(map #(vector (.getNodeId %) (.getPort %))))
all-executors (->> topology
.getExecutors
(map #(vector (.getStartTask %) (.getEndTask %)))
set)
alive-assigned (EvenScheduler/get-alive-assigned-node+port->executors cluster topology-id)
alive-executors (->> alive-assigned vals (apply concat) set)
can-reassign-slots (slots-can-reassign cluster (keys alive-assigned))
total-slots-to-use (min (.getNumWorkers topology)
(+ (count can-reassign-slots) (count available-slots)))
bad-slots (if (or (> total-slots-to-use (count alive-assigned))
(not= alive-executors all-executors))
(bad-slots alive-assigned (count all-executors) total-slots-to-use)
[])]]
(.freeSlots cluster bad-slots)
(EvenScheduler/schedule-topologies-evenly (Topologies. {topology-id topology}) cluster))))

1,取出需要scheduling的topologies (cluster.needsScheduling)

判断是否需要scheduling, 满足(or)
现在已经assigned的worker数小于配置的worker数 (dead slot或上次分配是可用slot不够)
all executors > assigned executors(新的topology或已分配的executor dead(没有hb))

public boolean needsScheduling(TopologyDetails topology) {
int desiredNumWorkers = topology.getNumWorkers();
int assignedNumWorkers = this.getAssignedNumWorkers(topology);
if (desiredNumWorkers > assignedNumWorkers) {
return true;
}
return this.getUnassignedExecutors(topology).size() > 0;
}

2, 对于每个需要scheduling的topology

2.1 找出cluster中所有可用的slots, 从每个SupervisorDetails中读出可用的slots(即assignedports - usedPorts)

available-slots, ([node1 port2] [node2 port2])

2.2 读出该topology所有的executors

all-executors , ([1 3] [4 4] [5 7])

2.3 从cluster中读出该topology的assignment关系

因为前面只将alive executors的assignment关系记录到cluster中, 所以从alive-assigned可用推出alive-executors

alive-assigned, node+port->executor, {[node1 port1] [1 3], [node2 port1] [5 7]}

alive-executors, ([1 3] [5 7])

2.4 找出topology当前运行的slots中哪些是可用的 (slots-can-reassign)

alive executors是有可能跑在dead slot上的, 所以不是所有alive executors的slot都可用

reassign的条件, node不在cluster的blacklist, port是否在supervisor的allPort中(即不是dead port), 即这个slot是可用的

可用的slot, 就可以用于reassign

2.5 total-slots-to-use应该等于(available-slots + can-reassign-slots)

当然最多slots数不能大于topology配置的worker number, 在可用slot数不够的情况下, 可能小于

2.6找出bad slots

针对不合理或bad的slots assignment关系, 找出相应的slots

并在下一步释放掉这些不合理的slots assignment

一般两种情况, 前一次分配时可用slots不够, 所以没有达到配置的数目;使用中某slot dead, 导致alive slot减少

if (or (> total-slots-to-use (count alive-assigned));;当前可用slots数大于当前assign的slots数

(not= alive-executors all-executors));;某些executors死了, 表明肯定有坏的slots, 或新的topology, 还没有分配
(bad-slots alive-assigned (count all-executors) total-slots-to-use) [])

 

(defn- bad-slots [existing-slots num-executors num-workers]
(if (= 0 num-workers)
'()
(let [distribution (atom (integer-divided num-executors num-workers))
keepers (atom {})]
(doseq [[node+port executor-list] existing-slots :let [executor-count (count executor-list)]]
(when (pos? (get @distribution executor-count 0)) ;;是否在正常的distribution中可以找到, 找到说明这个slot分配合理, 需要keep
(swap! keepers assoc node+port executor-list) ;;slot分配合理, 所以加到keeper中
(swap! distribution update-in [executor-count] dec) ;;并且该分配比例的份数减一
))
(->> @keepers
keys
(apply dissoc existing-slots) ;;在exisiting-slots中除去keeper, 剩下的都是bad-slots
keys
(map (fn [[node port]]
(WorkerSlot. node port)))))))

例子, 7个executor, 3个worker, 那么正常情况下, ((2 2)(3 1)), 1份3个, 2份2个

所以check所有现有的assignment, 把符合正常分配比例的加到keeper上, 比如这个case如果出现1个或2份3个都是不符合比例的

这些slot都被认为是bad slots

 

3 free bad slots

所谓的free, 就是在SchedulerAssignmentImpl中, 把所有bad slot上的executors从executorToSlot中删除

slot只要没有executor占用就是free

Map<ExecutorDetails, WorkerSlot> executorToSlot;
/**
* Release the slot occupied by this assignment.
* @param slot
*/
public void unassignBySlot(WorkerSlot slot) {
List<ExecutorDetails> executors = new ArrayList<ExecutorDetails>();
for (ExecutorDetails executor : this.executorToSlot.keySet()) {
WorkerSlot ws = this.executorToSlot.get(executor);
if (ws.equals(slot)) {
executors.add(executor);
}
}
// remove
for (ExecutorDetails executor : executors) {
this.executorToSlot.remove(executor);
}
}

4 EvenScheduler/schedule-topologies-evenly

这个function是doseq的经典应用, 两层doseq的嵌套

第一层doseq的处理函数, 仍然是一个doseq

第二层doseq的处理函数, .assign

(defn schedule-topologies-evenly [^Topologies topologies ^Cluster cluster]
(let [needs-scheduling-topologies (.needsSchedulingTopologies cluster topologies)]
(doseq [^TopologyDetails topology needs-scheduling-topologies
:let [topology-id (.getId topology)
new-assignment (schedule-topology topology cluster)
node+port->executors (reverse-map new-assignment)]]
(doseq [[node+port executors] node+port->executors
:let [^WorkerSlot slot (WorkerSlot. (first node+port) (last node+port))
executors (for [[start-task end-task] executors]
(ExecutorDetails. start-task end-task))]]
(.assign cluster slot topology-id executors)))))

4.1 调用schedule-topology

(defn- schedule-topology [^TopologyDetails topology ^Cluster cluster]
(let [topology-id (.getId topology)
available-slots (->> (.getAvailableSlots cluster)
(map #(vector (.getNodeId %) (.getPort %))))
all-executors (->> topology
.getExecutors
(map #(vector (.getStartTask %) (.getEndTask %)))
set)
alive-assigned (get-alive-assigned-node+port->executors cluster topology-id) ;;必须重新计算, 因为刚刚free的slot assignment关系
total-slots-to-use (min (.getNumWorkers topology)
(+ (count available-slots) (count alive-assigned)))
reassign-slots (take (- total-slots-to-use (count alive-assigned)) ;;上一步已经把bad slot都free, 仍然alive说明改slot不需要被reassign
(sort-slots available-slots))
reassign-executors (sort (set/difference all-executors (set (apply concat (vals alive-assigned)))))
reassignment (into {}
(map vector
reassign-executors
;; for some reason it goes into infinite loop without limiting the repeat-seq
(repeat-seq (count reassign-executors) reassign-slots)))]
(when-not (empty? reassignment)
(log-message "Available slots: " (pr-str available-slots))
)
reassignment))

 

reassign-slots,

a. 计算现在可以用于assignment的slots数, 之所以不直接使用available-slots, 因为有worker number限制, 所以可能小于available-slots

b. sort-slots, 按照port对slots进行排序

(defn sort-slots [all-slots] ;'(["n1" "p1"] ["n1" "p2"] ["n1" "p3"] ["n2" "p1"] ["n3" "p1"] ["n3" "p2"])
(let [split-up (vals (group-by first all-slots))]
(apply interleave-all split-up) ;'(["n1" "p1"] ["n2" "p1"] ["n3" "p1"] ["n1" "p2"] ["n3" "p2"] ["n1" "p3"])
))
c.从排过序的list取前(a.)个slots, 之所以前面按port排序, 可以使executors尽量分布在不同的node上

reassign-executors, 现在没有被assign的executors

assignment的过程非常简单, 就是对reassign-executors 和 reassign-slots做map, 注释中解释为什么要加count, 其实应该是不用加的, 因为理论上是当一个coll结束就会停止, 但某种原因似乎这里会停不下来.

之所以需要repeat-seq, 是因为executors往往多于slots

(map vector #{[1,3] [4,4] [5,6]}(repeat-seq 3 '(["n1" "p1"] ["n1" "p2"])))
([[4 4] ["n1" "p1"]] [[5 6] ["n1" "p2"]] [[1 3] ["n1" "p1"]])

4.2 将新的assignment封装成WorkerSlot和ExecutorDetails

4.3 最终将新的assignment结果放到SchedulerAssignmentImpl的executorToSlot中去

/**
* Assign the slot to executors.
* @param slot
* @param executors
*/
public void assign(WorkerSlot slot, Collection<ExecutorDetails> executors) {
for (ExecutorDetails executor : executors) {
this.executorToSlot.put(executor, slot);
}
}