2019独角兽企业重金招聘Python工程师标准>>> 
动态分配executor的实例初始化部分
如果spark.executor.instances配置项设置为0或者没有设置,这个默认情况下是一个未设置的值,yarn的运行模式时,这个配置通过--num-executors来得到.
同时spark.dynamicAllocation.enabled配置项设置为true时.默认值为false,表示启用了动态分配executor.
在driver端SparkContext生成时,会检查上面两个配置项,如果这两个配置满足动态executor分配的要求时,会生成一个ExecutorAllocationManager实例.
_executorAllocationManager =
if (dynamicAllocationEnabled) {
Some(new ExecutorAllocationManager(this, listenerBus, _conf))
} else {
None
}
_executorAllocationManager.foreach(_.start())
必要的配置项:
1,配置项spark.dynamicAllocation.minExecutors,默认值0,最少分配的executor的个数.
2,配置项spark.dynamicAllocation.maxExecutors,默认值int.maxvalue.最大可分配的executor的个数.
3,配置项spark.dynamicAllocation.initialExecutors,默认值为配置项1的值,初始化时启用的executor的个数,
4,1,配置项spark.dynamicAllocation.schedulerBacklogTimeout,默认值1s,如果未分配的task等待分配的时间超过了这个配置的时间,表示需要新启动executor.
4,2,配置项spark.dynamicAllocation.sustainedSchedulerBacklogTimeout,默认是4,1,配置项的值,这个配置用于设置在初始调度的executor调度延时后,每次的等待超时时间.
5,配置项spark.dynamicAllocation.executorIdleTimeout,默认值60s,executor的空闲回收时间.
6,配置项spark.executor.cores的配置(executor-cores)必须大于或等于配置项spark.task.cpus的值(这个配置默认是1,这是每个task需要的cpu的个数).
7,配置项spark.shuffle.service.enabled必须配置为true,默认为false.如果这个配置设置为true时,BlockManager实例生成时,需要读取spark.shuffle.service.port配置项配置的shuffle的端口,同时对应BlockManager的shuffleClient不在是默认的BlockTransferService实例,而是ExternalShuffleClient实例.
8,初始化时,ExecutorAllocationManager中的属性initializing默认值为true,表示定时调度时,什么都不做.
在执行ExecutorAllocationManager中的start函数时:
def start(): Unit = {
这里把ExecutorAllocationListener实例(内部实现类)添加到sparkContext中的listenerBus中,用于监听stage,task的启动与完成,并做对应的操作.
listenerBus.addListener(listener)
val scheduleTask = new Runnable() {
override def run(): Unit = {
try {
schedule()
} catch {
case ct: ControlThrowable =>
throw ct
case t: Throwable =>
logWarning(s"Uncaught exception in thread
${Thread.currentThread().getName}", t)
}
}
}
定时100ms执行一次schedule的调度函数,来进行task的分析.
executor.scheduleAtFixedRate(scheduleTask, 0, intervalMillis,
TimeUnit.MILLISECONDS)
}
对executor的个数分配的计算
针对task的调度主要由一个定时器每100ms进行一次schedule函数的调用.
private def schedule(): Unit = synchronized {
在这个函数中,首先得到当前的时间,
val now = clock.getTimeMillis
在调用这个函数时,初始情况下,initializing的属性值为true,这个时候,这个函数什么也不做.
这个函数的内容,后面在进行分析.
updateAndSyncNumExecutorsTarget(now)
这个removeTimes集合中记录有每一个executor没有被task占用后的时间,如果这个时间超过了上面配置的idle的时间,会移出掉这个executor,同时设置initializing属性为false,表示可以继续进行task的调度.retain函数只保留未超时的executor.
removeTimes.retain { case (executorId, expireTime) =>
val expired = now >= expireTime
if (expired) {
initializing = false
removeExecutor(executorId)
}
!expired
}
}
如何知道stage被提交?看下面,
在SparkContext中,执行runJob命令时,针对一个stage进行submit操作时,会调用listenerBus中所有的listener对应的onStageSubmitted函数.
而在ExecutorAllocationManager进行start操作时,生成了一个listener,实例为ExecutorAllocationListener,并把这个listener添加到了listenerBus中.
接下来看看ExecutorAllocationListener中对应stage提交的监听处理:
override def onStageSubmitted(stageSubmitted: SparkListenerStageSubmitted)
: Unit = {
这里首先把initializing的属性值设置为false,表示下次定时调度时,需要执行executor的分配操作.
initializing = false
得到进行submit操作对应的stage的id与stage中对应的task的个数.
val stageId = stageSubmitted.stageInfo.stageId
val numTasks = stageSubmitted.stageInfo.numTasks
allocationManager.synchronized {
通过对应的stageId设置这个stage的task的个数,存储到stageIdToNumTasks集合中.
stageIdToNumTasks(stageId) = numTasks
这里更新allocationManager中的addTime的时间,
由当前时间加上配置spark.dynamicAllocation.schedulerBacklogTimeout的超时时间.
allocationManager.onSchedulerBacklogged()
这里根据每个task对应的host,计算出每个host对应的task的个数,numTasksPending的个数原则上应该与stage中numTask的个数相同.
// Compute the number of tasks requested by the stage on each host
var numTasksPending = 0
val hostToLocalTaskCountPerStage = new mutable.HashMap[String, Int]()
stageSubmitted.stageInfo.taskLocalityPreferences.foreach { locality =>
if (!locality.isEmpty) {
numTasksPending += 1
locality.foreach { location =>
val count = hostToLocalTaskCountPerStage.getOrElse(location.host, 0) + 1
hostToLocalTaskCountPerStage(location.host) = count
}
}
}
在对应的集合中,根据stageid与pending的task的个数,对应的host与host对应的task的个数进行存储.
stageIdToExecutorPlacementHints.put(stageId,
(numTasksPending, hostToLocalTaskCountPerStage.toMap))
下面的函数迭代stageIdToExecutorPlacementHints集合中的values,并更新allocationManager中localityAwareTasks属性(存储待启动的task的个数)与hostToLocalTaskCount集合属性(存储host对应的task的个数)的值.添加到这里,主要是executor启动时对应的调度启动task
// Update the executor placement hints
updateExecutorPlacementHints()
}
}
接面看看allocationManager中定时调度的updateAndSyncNumExecutorsTarget函数:
现在来说说updateAndSyncNumExecutorsTarget函数与addExecutors函数的作用:
示例说明:
假定这次的stage需要的executor的个数为5,numExecutorsTarget的配置保持默认值0,
如果是第一次调度启动时,在updateAndSyncNumExecutorsTarget函数中:
1,先计算出这个stage需要的executor的个数,
val maxNeeded = maxNumExecutorsNeeded
if (initializing) {
如果函数进行这里,表示还没有stage提交,也就是没有job被执行.不进行调度.
// Do not change our target while we are still initializing,
// Otherwise the first job may have to ramp up unnecessarily
0
}
2,进入的流程为else if (addTime != NOT_SET && now >= addTime)部分.这个时候执行addExecutors函数,(这里假定时间已经达到了addTime的超时时间)
这种情况下默认的初始executor的个数为0的情况下,在当前时间超过了等待超时时间后,会进入,第一次时需要等待一秒钟,每次执行会更新等待时间.
这里根据要stage对应的task需要的executor的个数,并执行addExecutors的函数.
else if (addTime != NOT_SET && now >= addTime) {
val delta = addExecutors(maxNeeded)
logDebug(s"Starting timer to add more executors (to " +
s"expire in $sustainedSchedulerBacklogTimeoutS seconds)")
addTime += sustainedSchedulerBacklogTimeoutS * 1000
delta
}
在addExecutors函数中,先计算出目标的executor的个数(属性numExecutorsTarget),
// Do not request more executors if it would put our target over the upper bound
if (numExecutorsTarget >= maxNumExecutors) {
logDebug(s"Not adding executors because our current target total " +
s"is already $numExecutorsTarget (limit $maxNumExecutors)")
numExecutorsToAdd = 1
return 0
}
val oldNumExecutorsTarget = numExecutorsTarget
// There's no point in wasting time ramping up to the number of executors we already have, so
// make sure our target is at least as much as our current allocation:
numExecutorsTarget = math.max(numExecutorsTarget, executorIds.size)
// Boost our target with the number to add for this round:
numExecutorsTarget += numExecutorsToAdd
// Ensure that our target doesn't exceed what we need at the present moment:
numExecutorsTarget = math.min(numExecutorsTarget, maxNumExecutorsNeeded)
// Ensure that our target fits within configured bounds:
numExecutorsTarget = math.max(math.min(numExecutorsTarget, maxNumExecutors),
minNumExecutors)
此时executorIds的长度为0,集合是个空集合,这个时候numExecutorsToAdd的值为默认的1,根据上面的代码计算完成后(maxNumExecutorsNeeded为5就是tasks需要的executor的个数),numExecutorsTarget的值为1.接下来计算出来一个值,如果这次任务的目标executor的个数高于上次tasks启动的目标executor的个数,delta的值是一个大于0的值.根据上面的说明,下面代码中这个delta的值为1,
val delta = numExecutorsTarget - oldNumExecutorsTarget
// If our target has not changed, do not send a message
// to the cluster manager and reset our exponential growth
if (delta == 0) {
如果delta等于0,表示这次的目标executor的个数,与上次任务的executor的个数相同,重置增量的个数为1.
numExecutorsToAdd = 1
return 0
}
接下来,通过下面的代码通过SparkContext发起numExecutorsTarget的executor的启动,并在executor中加载对应的task的个数.
val addRequestAcknowledged = testing ||
client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks,
hostToLocalTaskCount)
接下来,由于我们的任务执行还需要的executor的个数还需要4个(共需要),同时这个时候,delta的值为1,与numExecutorsToAdd的属性值相同,因此numExecutorsToAdd的值会*2.
numExecutorsToAdd = if (delta == numExecutorsToAdd) {
numExecutorsToAdd * 2
} else {
1
}
3,调度定时器开始执行第二次调度启动,这个时候执行updateAndSyncNumExecutorsTarget函数时,numExecutorsTarget的值为1,需要的executor的个数为3,因此,还是会执行时间超时的流程.
else if (addTime != NOT_SET && now >= addTime) {
val delta = addExecutors(maxNeeded)
logDebug(s"Starting timer to add more executors (to " +
s"expire in $sustainedSchedulerBacklogTimeoutS seconds)")
addTime += sustainedSchedulerBacklogTimeoutS * 1000
delta
}
再次进入addExecutors函数,这个时候numExecutorsToAdd属性值为2,numExecutorsTarget属性值为1,executorsIds的size为1,已经有一个executor被启动,需要的executor的个数为3,最后计算完成后,numExecutorsTarget属性的值为3.计算出来当前的numExecutorsTarget与上一次的numExecutorsTarget的delta的值为2,开始根据这个值为3的numExecutorsTarget发起task的启动请求.
接下来,由于计算出来的delta的值为2,而属性numExecutorsToAdd的值也为2,
因此numExecutorsToAdd属性值现在还是需要*2操作.执行完成后,最后这个numExecutorsToAdd属性值修改成了4.
4,这个时候,由于还有部分task没有被执行,开始第三次的处理,此时,numExecutorTarget的值还是小于目标的executor的个数,接着执行addExecutors函数,此时,executorsIds的size为4,第一次执行一个,第二次启动了3个,这个时候,numExecutorsTarget的属性值变化情况:
首先先修改成4,(取numExecutorsTarget与executorIds.size中的最大值),
然后numExecutorsTarget += numExecutorsToAdd的值,这个时候值修改成了8.
最后与共需要的executor的个数5取最小值,把值修改成5.计算出当前的numExecutorsTarget与上一次的numExecutorsTarget的差值为2,numExecutorsToAdd的值为4,因此重新修改numExecutorsToAdd的值为1.
这个时候调度程序会修改addTime的值为NOT_SET,表示不在执行executor的调度.因为executor已经够了.
5,现在假定spark.dynamicAllocation.initialExecutors配置项配置有一个值,初始值为6.需要的executor的个数还是是5.这个时候,进入updateAndSyncNumExecutorsTarget函数时,执行如下的流程部分,因为初始的executor的个数大于了需要的executor的个数.这部分流程在设置有初始大小的executor个数或者说要执行的job的第二个stage的task的个数需要的executor的个数小于小次stage需要的executor的个数时,会被执行.
else if (maxNeeded < numExecutorsTarget) {
// The target number exceeds the number we actually need, so stop adding new
// executors and inform the cluster manager to cancel the extra pending requests
val oldNumExecutorsTarget = numExecutorsTarget
numExecutorsTarget = math.max(maxNeeded, minNumExecutors)
numExecutorsToAdd = 1
// If the new target has not changed, avoid sending a message to the cluster manager
if (numExecutorsTarget < oldNumExecutorsTarget) {
client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks,
hostToLocalTaskCount)
logDebug(s"Lowering target number of executors to $numExecutorsTarget
(previously " +
s"$oldNumExecutorsTarget) because not all requested executors are actually
needed")
}
numExecutorsTarget - oldNumExecutorsTarget
}
在上面的代码中,重新根据需要的executor的个数,计算出numExecutorsTarget的值,这个时候,新的numExecutorsTarget的值为5,而老的numExecutorsTarget的值为6,因此通过新的numExecutorsTarget直接调用SparkContext中对应的启动executor的函数,发起对executor的调度与task的启动.
通过SparkContext调度executor
在allocationManager中,对executor进行动态的调用后,会执行如下的代码片断.
client.requestTotalExecutors(numExecutorsTarget, localityAwareTasks,
hostToLocalTaskCount)
在上面的代码中,client就是SparkContext实例.
下面看看这个函数的处理流程:
函数的传入参数中:
numExecutors是目标的executor的个数,
第二个是共需要的task的个数,
第三个是host->taskCount的集合.
private[spark] override def requestTotalExecutors(
numExecutors: Int,
localityAwareTasks: Int,
hostToLocalTaskCount: scala.collection.immutable.Map[String, Int]
): Boolean = {
schedulerBackend match {
case b: CoarseGrainedSchedulerBackend =>
这里直接调用了GoarseGrainedSchedulerBackend中对应的函数.
b.requestTotalExecutors(numExecutors, localityAwareTasks,
hostToLocalTaskCount)
case _ =>
logWarning("Requesting executors is only supported in coarse-grained mode")
false
}
}
下面看看GoarseGrainedSchedulerBackend中requestTotalExecutors的函数实现:
final override def requestTotalExecutors(
numExecutors: Int,
localityAwareTasks: Int,
hostToLocalTaskCount: Map[String, Int]
): Boolean = synchronized {
if (numExecutors < 0) {
throw new IllegalArgumentException(
"Attempted to request a negative number of executor(s) " +
s"$numExecutors from the cluster manager. Please specify a positive number!")
}
this.localityAwareTasks = localityAwareTasks
this.hostToLocalTaskCount = hostToLocalTaskCount
每次执行时,计算出还需要的共需要的executor的个数与正在执行或者等待回收的executor的个数之间的差值,这个差值是还需要启动的executor的个数.
numPendingExecutors =
math.max(numExecutors - numExistingExecutors + executorsPendingToRemove.size,
0)
这里根据具体的cluster的部署模式(yarn,standalone,mesos,等),调用对应的函数进行executor的启动操作.这里我们看看standalone的操作.由SparkDeploySchedulerBackend实现.
这个函数的实现主要是通过向master发送一个RequestExecutors消息,这个消息是一个需要响应的消息.
这个消息在Master中通过receiveAndReply函数中的RequestExecutors部分进行处理.
doRequestTotalExecutors(numExecutors)
}
Master中处理executor的申请:
case RequestExecutors(appId, requestedTotal) =>
context.reply(handleRequestExecutors(appId, requestedTotal))
看看这个的handleRequestExecutors函数
private def handleRequestExecutors(appId: String, requestedTotal: Int)
: Boolean = {
idToApp.get(appId) match {
这个函数中,根据传入的app对应的job共依赖的executor的个数,更新appInfo中executorLimit的值.并执行对executor的启动的调度.
case Some(appInfo) =>
logInfo(s"Application $appId requested to set total executors to
$requestedTotal.")
appInfo.executorLimit = requestedTotal
在这个调度的过程中通过startExecutorsOnWorkers函数来调度与启动executor在对应的worker中,
在判断要启动的executor的个数时,会根据scheduleExecutorsOnWorkers函数来判断executor的个数是否达到要求的appInfo.executorLimit的个数,如果达到指定的executor的个数时,调度不再执行executor的启动.判断worker是否有足够的资源启动executor时,通过对executor需要的cpu core的个数与executor需要的内存来判断worker是否有足够的对应资源启动executor,如果有,表示这个worker可以用来启动executor,迭代所有的worker进行executor的启动,当已经启动的executor的个数达到了appInfo的executorLimit的限制时,不在进行分配.
schedule()
true
case None =>
logWarning(s"Unknown application $appId requested
$requestedTotal total executors.")
false
}
}
Driver端处理worker对executor的启动后的监听:
Worker端的executor启动后,会向driver发起一个RegisterExecutor消息.
这个消息在Driver端通过CoarseGrainedSchedulerBackend实例中的receiveAndReply函数中的RegisterExecutor部分来进行处理.
case RegisterExecutor(executorId, executorRef, hostPort, cores, logUrls) =>
if (executorDataMap.contains(executorId)) {
context.reply(RegisterExecutorFailed("Duplicate executor ID: " + executorId))
} else {
// If the executor's rpc env is not listening for incoming connections, `hostPort`
// will be null, and the client connection should be used to contact the executor.
val executorAddress = if (executorRef.address != null) {
executorRef.address
} else {
context.senderAddress
}
logInfo(s"Registered executor $executorRef ($executorAddress) with ID
$executorId")
这里把executor的信息存储到driver中对应的集合容器中.
并更新待启动的executor的个数(numPendingExecutors属性).
addressToExecutorId(executorAddress) = executorId
totalCoreCount.addAndGet(cores)
totalRegisteredExecutors.addAndGet(1)
val data = new ExecutorData(executorRef, executorRef.address,
executorAddress.host,
cores, cores, logUrls)
// This must be synchronized because variables mutated
// in this block are read when requesting executors
CoarseGrainedSchedulerBackend.this.synchronized {
executorDataMap.put(executorId, data)
if (numPendingExecutors > 0) {
numPendingExecutors -= 1
logDebug(s"Decremented number of pending executors
($numPendingExecutors left)")
}
}
// Note: some tests expect the reply to come after we put the executor in the map
context.reply(RegisteredExecutor(executorAddress.host))
这里通过向allocationManager中的ExecutorAllocationListener中executorAdded的监听函数.
listenerBus.post(
SparkListenerExecutorAdded(System.currentTimeMillis(), executorId, data))
这里在启动后的executors中对task进行调度与启动.
makeOffers()
}
在ExecutorAllocationManager中,对executor注册,task启动与结束的监听处理:
对executor的注册处理:
private def onExecutorAdded(executorId: String): Unit = synchronized {
if (!executorIds.contains(executorId)) {
把新启动的executor对应的id存储到 executorIds的集合中.
executorIds.add(executorId)
这里检查下所有的executor是否包含有task在运行,如果不包含的executor,添加到remoteTimes的集合中,这个remoteTimes中对应此executor有一个超时的清理时间,
如果executor不包含blockCache,
超时清理时间通过spark.dynamicAllocation.executorIdleTimeout配置.默认为60s
如果executor中包含有blockCache时,
超时清理时间通过spark.dynamicAllocation.cachedExecutorIdleTimeout配置.默认不控制超时.
// If an executor (call this executor X) is not removed because the lower bound
// has been reached, it will no longer be marked as idle. When new executors join,
// however, we are no longer at the lower bound,
and so we must mark executor X
// as idle again so as not to forget that it is a candidate for removal.
(see SPARK-4951)
executorIds.filter(listener.isExecutorIdle).foreach(onExecutorIdle)
logInfo(s"New executor $executorId has registered (new total is
${executorIds.size})")
} else {
logWarning(s"Duplicate executor $executorId has registered")
}
}
对task启动时的监听:
override def onTaskStart(taskStart: SparkListenerTaskStart): Unit = {
val stageId = taskStart.stageId
val taskId = taskStart.taskInfo.taskId
val taskIndex = taskStart.taskInfo.index
val executorId = taskStart.taskInfo.executorId
allocationManager.synchronized {
首先更新正在运行的task的个数,
numRunningTasks += 1
// This guards against the race condition in which the `SparkListenerTaskStart`
// event is posted before the `SparkListenerBlockManagerAdded` event, which is
// possible because these events are posted in different threads. (see SPARK-4951)
if (!allocationManager.executorIds.contains(executorId)) {
allocationManager.onExecutorAdded(executorId)
}
在stage与task的对应关系的集合中,设置stageId中包含的taskId的值.
// If this is the last pending task, mark the scheduler queue as empty
stageIdToTaskIndices.getOrElseUpdate(stageId, new mutable.HashSet[Int]) +=
taskIndex
这里检查的stageid中对应的所有的正在运行的task的个数是否达到了stage需要运行的task的个数,
如果达到了需要运行的task的个数,设置allocationManager中addTime属性的值为NOT_SET.同时重置属性numExecutorsToAdd的值为1.
if (totalPendingTasks() == 0) {
allocationManager.onSchedulerQueueEmpty()
}
把task运行的executor的对应关系存储到对应的集合中,这个存储的集合用于判断executor是否是闲置的状态.
// Mark the executor on which this task is scheduled as busy
executorIdToTaskIds.getOrElseUpdate(executorId, new mutable.HashSet[Long]) +=
taskId
由于executorId已经有task在运行,把这个executor从要移出的executor集合中移出
从removeTimes集合中移出,这个集合中存储了executor的闲置回收超时时间.
从executorsPendingToRemove集合中移出,这个集合中存储了准备执行移出操作的executor的集合.
allocationManager.onExecutorBusy(executorId)
}
}
对task停止时的监听:
override def onTaskEnd(taskEnd: SparkListenerTaskEnd): Unit = {
val executorId = taskEnd.taskInfo.executorId
val taskId = taskEnd.taskInfo.taskId
val taskIndex = taskEnd.taskInfo.index
val stageId = taskEnd.stageId
allocationManager.synchronized {
把对应的运行的TASK的个数减去1,表示有一个task完成运行.
numRunningTasks -= 1
从对应的executor的关系集合中移出此task,如果这个executor已经没有了正在运行的task时,直接把这个executor执行idle的处理,也就是移动到removeTimes集合中.
// If the executor is no longer running any scheduled tasks, mark it as idle
if (executorIdToTaskIds.contains(executorId)) {
executorIdToTaskIds(executorId) -= taskId
if (executorIdToTaskIds(executorId).isEmpty) {
executorIdToTaskIds -= executorId
allocationManager.onExecutorIdle(executorId)
}
}
// If the task failed, we expect it to be resubmitted later. To ensure we have
// enough resources to run the resubmitted task, we need to mark the scheduler
// as backlogged again if it's not already marked as such (SPARK-8366)
if (taskEnd.reason != Success) {
if (totalPendingTasks() == 0) {
allocationManager.onSchedulerBacklogged()
}
stageIdToTaskIndices.get(stageId).foreach { _.remove(taskIndex) }
}
}
}
对stage停止时的监听:
override def onStageCompleted(stageCompleted: SparkListenerStageCompleted)
: Unit = {
val stageId = stageCompleted.stageInfo.stageId
allocationManager.synchronized {
从对应的集合中移出此stage.
stageIdToNumTasks -= stageId
stageIdToTaskIndices -= stageId
stageIdToExecutorPlacementHints -= stageId
更新allocationManager中对应各个job的stage中task与location的对应关系.
// Update the executor placement hints
updateExecutorPlacementHints()
所有的job都运行完成,暂停调度.
// If this is the last stage with pending tasks, mark the scheduler queue as empty
// This is needed in case the stage is aborted for any reason
if (stageIdToNumTasks.isEmpty) {
allocationManager.onSchedulerQueueEmpty()
if (numRunningTasks != 0) {
logWarning("No stages are running, but numRunningTasks != 0")
numRunningTasks = 0
}
}
}
}
在调度过程中发现executor超时需要移出时的处理:
private def removeExecutor(executorId: String): Boolean = synchronized {
// Do not kill the executor if we are not aware of it (should never happen)
if (!executorIds.contains(executorId)) {
logWarning(s"Attempted to remove unknown executor $executorId!")
return false
}
// Do not kill the executor again if it is already pending to be killed
(should never happen)
if (executorsPendingToRemove.contains(executorId)) {
logWarning(s"Attempted to remove executor $executorId " +
s"when it is already pending to be removed!")
return false
}
这里检查下现在存在的executor的个数是否小于了配置的最小的executor的个数,如果不小于才执行移出操作.
// Do not kill the executor if we have already reached the lower bound
val numExistingExecutors = executorIds.size - executorsPendingToRemove.size
if (numExistingExecutors - 1 < minNumExecutors) {
logDebug(s"Not removing idle executor $executorId because there are only " +
s"$numExistingExecutors executor(s) left (limit $minNumExecutors)")
return false
}
这里发起对executor的停止操作.并把这个发起停止后的executor添加到executorsPendingToRemov集合中.等待回调此allocationManager中listener中的onExecutorRemoved函数.
// Send a request to the backend to kill this executor
val removeRequestAcknowledged = testing || client.killExecutor(executorId)
if (removeRequestAcknowledged) {
logInfo(s"Removing executor $executorId because it has been idle for " +
s"$executorIdleTimeoutS seconds (new desired total will be
${numExistingExecutors - 1})")
executorsPendingToRemove.add(executorId)
true
} else {
logWarning(s"Unable to reach the cluster manager to kill executor $executorId!")
false
}
}
allocationManager中监听executor的移出操作:
override def onExecutorRemoved(executorRemoved: SparkListenerExecutorRemoved)
: Unit = {
allocationManager.onExecutorRemoved(executorRemoved.executorId)
}
其实这个函数就是从对应的集合中移出此executor.
private def onExecutorRemoved(executorId: String): Unit = synchronized {
if (executorIds.contains(executorId)) {
executorIds.remove(executorId)
removeTimes.remove(executorId)
logInfo(s"Existing executor $executorId has been removed (new total is
${executorIds.size})")
if (executorsPendingToRemove.contains(executorId)) {
executorsPendingToRemove.remove(executorId)
logDebug(s"Executor $executorId is no longer pending to " +
s"be removed (${executorsPendingToRemove.size} left)")
}
} else {
logWarning(s"Unknown executor $executorId has been removed!")
}
}