spark 动态资源分配dynamicAllocation

news2024/11/23 11:17:54

动态资源分配，主要是spark在运行中可以相对合理的分配资源。

初始申请的资源远超实际需要，减少executor
初始申请的资源比实际需要少很多，增多executor
Spark运行多个job，这些job所需资源有的多有的少，动态调整executor数量

ExecutorAllocationManager

ExecutorAllocationManager是在SparkContext初始化的时候创建的，创建后调用它的start方法。

initializing变量标记ExecutorAllocationManager是否可以进行动态调整。

addTime变量是添加新的executor的时间点

start

在start方法首先注册了两个listener：

ExecutorAllocationListener：通知给定的分配管理器何时添加和删除执行器。
ExecutorMonitor：执行器活动的监视器，用于检测空闲执行器。

定时调度每100ms执行一次schedule方法。
最后向更新集群发送所需executor的信息。

numExecutors：向集群申请的executor数量。集群不一定为了达到这个数量就启动或者杀死executor
localityAwareTasks：stage中具有局部首选项的任务数。这包括正在运行、挂起和已完成的任务。有些task是有指定在哪里运行或者哪里不运行的。
hostToLocalTaskCount：host和希望在host上运行的task数量。包括正在运行、挂起和已完成的任务。

schedule

调用executorMonitor的timedOutExecutors获取超时的executor。
如有超时的executor，表明executor首次部署成功过，将initializing置为false，标志可以进行动态调整executor数量。
调用updateAndSyncNumExecutorsTarget方法向集群同步executor调度的相关信息，集群收到新的信息后会判断是否满足需求，不满足的话会添加executor。这里集群只可能增加executor来满足目标数量，不会进行kill executor。
最后调用removeExecutors移除超时的executor集合。

updateAndSyncNumExecutorsTarget

首先是调用maxNumExecutorsNeeded方法获取所需executor的最大数量。

initializing为true，表明executor首次还没有部署完成，不能动态调整
maxNeeded < numExecutorsTarget：此次所需的最大数量比上次申请的executor数量少，此时就要向集群更新executor目标数量，让集群可以停止还没有完成部署的executor的申请
addTime != NOT_SET && now >= addTime：到达添加时间，可以申请添加executor
其他情况：没有达到添加时间

maxNumExecutorsNeeded

计算当前任务所需要的最大executor数量。

addExecutors

计算新的executor目标值，每次新增都是加上numExecutorsToAdd变量值。再经过校验调整到合理的值。
如果跟上一次的目标值一致，表示新增executor过程完成了，重置numExecutorsToAdd为1。
向集群发送executor目标值，让集群根据情况调整。
最后调整numExecutorsToAdd方便下一次扩容。
executor新增的速度是 1 2 4 8…，这样做是因为新增速度为固定值会造成目标1.executor数量小，增长速度大，申请了过多的executor；2.目标executor数量大，增长速度小，executor扩容慢。

removeExecutors

移除executor不能直接将超时的executor都移除了，存活的executor数量还要大于等于executor最小数量、executor目标数量。
executorIdsToBeRemoved是实际需要移除的executor

向集群发送kill executor的命令，更新executor目标数量到集群。最后修改executorMonitor中对应executor状态为待移除，不再进行监控这些executor