之前阅读剖析过Spark Application注册的源码。

传送门：《Spark Application的注册源码剖析》

当时在看的过程中卡过很多点，其中如何Application是如何调度的，调度的算法如何做到最佳状态，和保持分布式的高效稳定的。今天于是带着问题去剖析一把。

Spark Application调度是一个庞大复杂的体系。

Application注册过程，在Application 注册源码剖析中，得知向Master发送注册消息。

RegisterApplication(description, driver):

判断Master状态
传入applicationDescription和driver，调用createApplication方法，创建application，得到封装的ApplicationInfo，即app
将app传入，注册application，将app加入内存缓存，将app加入等待队列
持久化引擎将app持久化
driver向StandaloneAppClient发送消息，application已注册，即RegisteredApplication
最后调用scheduler方法，安排调度

从createApplication开始，该方法根据将当前时间戳，ApplicationDescription,app id，日期，driver，默认cpu core数信息封装成了ApplicationInfo，并返回。

接收返回的ApplicationInfo，即app，接下来注册application调用registerApplication

首先判断driver地址是否存在，避免重复注册application
将app添加到apps HashSet中
将app id添加到idToApp HashMap中
将app driver添加到endpointToApp HashMap中
将app rpc地址添加到addressToApp HashMap中
将app添加到等待执行waitingApps ArrayBuffer中

接下来持久化引擎将app持久化

driver向StandaloneAppClient发送消息，application已注册，即RegisteredApplication

重点为最后调用的schedule( )方法

1.首先判断master的状态

2.调用startExecutorsOnWorkers方法

startExecutorsOnWorkers

先对worker资源进行筛选，cpu core数多的worker排在前面，将可用的worker添加到可用的usableWorkers集合中
调用scheduleExecutorsOnWorkers，为Application分配要在Worker上启动的Executor运行的Cpu Core数
分配好了executor 的core数
最后调用allocateWorkerResourceToExecutors，按照分配规则，分配worker资源给execut

OK快读一遍之后有个大致了解Application注册过程。其中schedule为具体调度。

每当集群资源发生变化时，active master 进程为所有已注册的并且没有调度完毕的application调度Worker节点上的Executor进程。而每个Executor上都分布了大量的task，Executor是一个进程，而task是线程中一个线程。在yarn资源管理器中,Executor中包含了大量的container。container就是资源，container包含了core,内存等资源信息 - -

理解了调度是干嘛了：继续去看调度算法。

经过查阅资料，Spark调度算法只有两种

spreadOut算法
非spreadOut算法

那么带着理解去直接追溯到master.scala类的schedule方法

打乱洗牌存活的Worker,在Worker资源满足的情况下，启动Executor。
（Driver资源调度）==> launchDriver(worker, driver)
（Executor调度）==> startExecutorsOnWorkers()

launchDriver：发送到Worker开始启动driver

Worker返回内容

startExecutorsOnWorkers：Schedule and launch executors on workers

coresPerExecutor：参数设置的每一个Executor所使用的内核数，默认为1。
app.desc.memoryPerExecutorMB ：参数设置的ExecutorMemory。
scheduleExecutorsOnWorkers ：返回各个Worker上分配的内核数

allocateWorkerResourceToExecutors：Allocate a worker's resources to one or more executors.将worker中资源分配给一个或者多个executors

Master 终端发送 LaunchExecutor

OK我自己看到有些一知半解的。。。。看到现在还没梳理spreadOut算法，非spreadOut算法。

翻阅资料，弄清楚了。以下博客园以为大佬写的博文非常清晰引用^[1]。

2.基本概念

每一个application至少包含以下基本属性:

coresPerExecutor：每一个Executor进程的cpu cores个数
memoryPerExecutor：每一个Executor进程的memory大小
maxCores: 这个application最多需要的cpu cores个数。

每一个worker至少包含以下基本属性：

freeCores:worker 节点当前可用的cpu cores个数
memoryFree:worker节点当前可用的memory大小。

假设一个待注册的application如下：

coresPerExecutor：2
memoryPerExecutor：512M
maxCores: 12

这表示这个application 最多需要12个cpu cores，每一个Executor进行都要2个core，512M内存。
假设某一时刻spark集群有如下几个worker节点,他们按照coresFree降序排列:

Worker1:coresFree=10  memoryFree=10G

Worker2:coresFree=7   memoryFree=1G

Worker3:coresFree=3   memoryFree=2G

Worker4:coresFree=2   memoryFree=215M

Worker5:coresFree=1   memoryFree=1G

其中worker5不满足application的要求：worker5.coresFree < application.coresPerExecutor
worker4也不满足application的要求:worker4.memoryFree < application.memoryPerExecutor
因此最终满足调度要求的worker节点只有前三个，我们将这三个节点记作usableWorkers。

3.SPREADOUT算法

先介绍spreadOut算法吧。上面已经说了，满足条件的worker只有前三个:

Worker1:coresFree=10  memoryFree=10G

Worker2:coresFree=7   memoryFree=1G

Worker3:coresFree=3   memoryFree=2G

第一次调度之后，worker列表如下:

Worker1:coresFree=8  memoryFree=9.5G  assignedExecutors=1  assignedCores=2

Worker2:coresFree=7  memoryFree=1G    assignedExecutors=0  assignedCores=0

Worker3:coresFree=3  memoryFree=2G    assignedExecutors=0  assignedCores=0

totalExecutors:1,totalCores=2

可以发现，worker1的coresFree和memoryFree都变小了而worker2，worker3并没有发生改变，这是因为我们在worker1上面分配了一个Executor进程(这个Executor进程占用2个cpu cores，512M memory)而没有在workre2和worker3上分配。

接下来继续循环，开始去worker2上分配:

Worker1:coresFree=8  memoryFree=9.5G      assignedExecutors=1  assignedCores=2

Worker2:coresFree=5  memoryFree=512M      assignedExecutors=1  assignedCores=2

Worker3:coresFree=3  memoryFree=2G        assignedExecutors=0  assignedCores=0

totalExecutors:2,totalCores=4

此时已经分配了2个Executor进程，4个core。

接下来去worker3上分配:

Worker1:coresFree=8  memoryFree=9.5G      assignedExecutors=1  assignedCores=2

Worker2:coresFree=5  memoryFree=512M      assignedExecutors=1  assignedCores=2

Worker3:coresFree=1  memoryFree=1.5G      assignedExecutors=1  assignedCores=2

totalExecutors:3,totalCores=6

接下来再去worker1分配，然后worker2...依此类推...以round-robin方式分配，由于worker3.coresFree < application.coresPerExecutor,不会在它上面分配资源了：

Worker1:coresFree=6  memoryFree=9.0G      assignedExecutors=2  assignedCores=4

Worker2:coresFree=5  memoryFree=512M      assignedExecutors=1  assignedCores=2

Worker3:coresFree=1  memoryFree=1.5G      assignedExecutors=1  assignedCores=2

totalExecutors:4,totalCores=8

Worker1:coresFree=6  memoryFree=9.0G      assignedExecutors=2  assignedCores=4

Worker2:coresFree=3  memoryFree=0M        assignedExecutors=2  assignedCores=4

Worker3:coresFree=1  memoryFree=1.5G      assignedExecutors=1  assignedCores=2

totalExecutors:5,totalCores=10

此时worker2也不满足要求了：worker2.memoryFree < application.memoryPerExecutor
因此，下一次分配就去worker1上了：

Worker1:coresFree=4  memoryFree=8.5G      assignedExecutors=3  assignedCores=6

Worker2:coresFree=3  memoryFree=0M        assignedExecutors=2  assignedCores=4

Worker3:coresFree=1  memoryFree=1.5G      assignedExecutors=1  assignedCores=2

totalExecutors:6,totalCores=12

ok，由于已经分配了12个core，达到了application的要求，所以不在为这个application调度了。

4.非SPREADOUT算法

那么非spraadOut算法呢？他是逮到一个worker如果不把他的资源耗尽了是不会放手的：

Worker1:coresFree=8  memoryFree=9.5G  assignedExecutors=1  assignedCores=2

Worker2:coresFree=7  memoryFree=1G    assignedExecutors=0  assignedCores=0

Worker3:coresFree=3  memoryFree=2G    assignedExecutors=0  assignedCores=0

totalExecutors:1,totalCores=2

Worker1:coresFree=6  memoryFree=9.0G  assignedExecutors=2  assignedCores=4

Worker2:coresFree=7  memoryFree=1G    assignedExecutors=0  assignedCores=0

Worker3:coresFree=3  memoryFree=2G    assignedExecutors=0  assignedCores=0

totalExecutors:2,totalCores=4

Worker1:coresFree=4  memoryFree=8.5    assignedExecutors=3  assignedCores=6

Worker2:coresFree=7  memoryFree=1G     assignedExecutors=0  assignedCores=0

Worker3:coresFree=3  memoryFree=2G     assignedExecutors=0  assignedCores=0

totalExecutors:3,totalCores=6

Worker1:coresFree=2   memoryFree=8.0G  assignedExecutors=4  assignedCores=8

Worker2:coresFree=7   memoryFree=1G    assignedExecutors=0  assignedCores=0

Worker3:coresFree=3   memoryFree=2G    assignedExecutors=0  assignedCores=0

totalExecutors:4,totalCores=8

Worker1:coresFree=0   memoryFree=7.5G  assignedExecutors=5  assignedCores=10

Worker2:coresFree=7   memoryFree=1G    assignedExecutors=0  assignedCores=0

Worker3:coresFree=3   memoryFree=2G    assignedExecutors=0  assignedCores=0

totalExecutors:5,totalCores=10

当worker1的coresfree已经耗尽了。由于application需要12个core，而这里才分配了10个，所以还要继续往下分配：

Worker1:coresFree=0   memoryFree=7.5G      assignedExecutors=5  assignedCores=10

Worker2:coresFree=5   memoryFree=512G      assignedExecutors=1  assignedCores=2

Worker3:coresFree=3   memoryFree=2G        assignedExecutors=0  assignedCores=0

totalExecutors:6,totalCores=12

ok，最终分配来12个core，满足了application的要求。
对比：

spreadOut算法中，是以round-robin方式，轮询的在worker节点上分配Executor进程，即以如下序列分配:worker1,worker2... ... worker n,worker1... .... worker n

非spreadOut算法中，逮者一个worker就不放手，直到满足一下条件之一：
worker.freeCores < application.coresPerExecutor 或者 worker.memoryFree<application.memoryPerExecutor。