Spark 动态资源失效问题排查

╰+哭是因爲堅強的太久メ 2022-04-24 07:46 552阅读 0赞

### 文章目录 ###

*   *  一、问题描述
     *  二、动态资源相关原理
     *   *  初试executor数量
         *  executor数量的变动
         *   *  1、Executor Add
             *  2、Executor Remove
     *  三、问题定位
     *   *  Spark Task的本地化调度
         *  本地化调度级别没有及时推进导致的问题
     *  四、解决方案
     *   *  1、问题一
         *  2、问题二
         *  3、总结

## 一、问题描述 ##

最近开启动态资源后，有用户反馈他的任务运行很慢，去Spark HistoryServer页面看了下，**发现只剩下一个executor在运行，而运行中的Stage还有几千个task等待执行**。executor数量并没有达到上限，但是spark却没有去申请更多的executor来运行task。

下面是该任务和动态资源有关的相关配置：

spark.dynamicAllocation.enabled true
    spark.executor.instances 60
    spark.dynamicAllocation.executorIdleTimeout 30s
    spark.dynamicAllocation.minExecutors 1
    spark.dynamicAllocation.maxExecutors 60

从配置可以看出，这个任务是开启了动态资源了，因此最多可以申请60个executor运行。观察日志发现一开始运行时也有60个executor在运行，但是随着任务的运行，executor最终只剩下一个（其他的executor都是因为空闲时间超过30s被kill）。

总结了下，一共有两个可疑的地方：

1.  看日志发现，executor因为空闲被remove时，还有很多待运行的task。为什么executor会空闲长达30s不接受任务运行呢？
2.  executor只剩下一个的时候，待运行的task数量也还有很多。那为什么没按照动态资源的规则继续申请新的executor呢？

## 二、动态资源相关原理 ##

### 初试executor数量 ###

无论是否开启动态资源，Spark Job在启动时都会去申请一定数量的executor。在开启动态资源时，这个数量由三个配置参数共同决定：

*  spark.executor.instances
 *  spark.dynamicAllocation.minExecutors
 *  spark.dynamicAllocation.initialExecutors

如果这三个值都设置了，**spark会取其中的最大值最为初始executor的数量**。

### executor数量的变动 ###

在动态资源开启后，executor的数量会随着程序的运行进行增加。动态资源的管理主要由ExecutorAllocationManager类来负责，只有将spark.dynamicAllocation.enabled设置为true，ExecutorAllocationManager类才会初始化并启动。

下面是ExecutorAllocationManager类中一些和executor数量变化有关的参数：

//最少executor数量
    int minNumExecutors
    //最多executor数量
    int maxNumExecutors
    //目前期望的executor数量，这个的取值会不断变动，一般根据 (正在等待运行的task数量 + 运行中的task数量) 来决定
    int numExecutorsTarget = initialNumExecutors
    //要申请executor时递增的数量，初始值为1，后面每次申请都会让该值翻倍，也就是慢慢变成 1 2 4 8 16 32 ...
    //当然，即使numExecutorsToAdd再大，申请的最终数量也不能超过maxNumExecutors
    int numExecutorsToAdd = 1

ExecutorAllocationManager类在启动时，会开启一个定时任务线程，线程名为"spark-dynamic-executor-allocation"。该任务每隔100ms运行一次。该任务主要做两件事：

1.  检查是否需要申请新的executor
2.  检查是否有executor空闲太久，需要kill掉

#### 1、Executor Add ####

是否需要申请新的executor和参数numExecutorsTarget密切相关。每次定时任务运行时，spark都会先获取需要executor的总数量，计算公式如下:

val numRunningOrPendingTasks = listener.totalPendingTasks + listener.totalRunningTasks
    (numRunningOrPendingTasks + tasksPerExecutor - 1) / tasksPerExecutor

tasksPerExecutor就是每个executor可以运行的task数量。这个值由参数`spark.executor.cores`和`spark.task.cpus`相除获得。  
获取到当前需要executor的总数量后，就拿这个值和numExecutorsTarget的值相比，如果需要的executor的总数量比numExecutorsTarget小，就说明此时executor数量太多，则主动减少executor数量，并更新numExecutorsTarget的值。  
如果该值比numExecutorsTarget多，就可以尝试去申请新的executor。申请前还需要判断numExecutorsTarget是否已经大于等于maxNumExecutors，如果是则不能再申请新的executor。  
申请的数量主要和参数numExecutorsToAdd有关，这个参数从1开始。每次申请完，这个参数都会翻倍，也就是1、2、4、8、16、…。另外，申请的时候也会判断申请的数量是否超过了maxNumExecutors来保证不超过maxNumExecutors。

下面列出整个流程的java伪代码：

//获取当前需要的executor数量
    int maxNumExecutorsNeeded = getExecutorsNeeded()
    if(maxNumExecutorsNeeded < numExecutorsTarget){ 
        //通知集群当前需要的executor数量
        requestTotalExecutors(maxNumExecutorsNeeded)
        numExecutorsTarget = maxNumExecutorsNeeded
    }else{ 
        if (numExecutorsTarget >= maxNumExecutors){ 
            //不申请任何executor就退出
            return
        }else{ 
            int newTargetExecutor = mint(numExecutorsTarget + numExecutorsToAdd,maxNumExecutors)
            //通知集群当前需要的executor数量
            requestTotalExecutors(newTargetExecutor)
            numExecutorsTarget = maxNumExecutorsNeeded
            numExecutorsToAdd = numExecutorsToAdd * 2
        }
    }

**注意: requestTotalExecutors方法的入参不是说要再申请几个，而是告诉集群目标需要几个executor。如果这个值比集群目前运行的executor数量还少，集群也不会为了达到这个值强行kill掉一些executor。因此，调用这个方法并不会导致executor数量的减少。**

#### 2、Executor Remove ####

在开启动态资源后，对于那些空闲太久的executor，ExecutorAllocationManager类会主动kill掉这个executor来释放资源。

ExecutorAllocationManager类会维护每个executor的过期时间，这个过期时间由参数`spark.dynamicAllocation.executorIdleTimeout`或者`spark.dynamicAllocation.cachedExecutorIdleTimeout`决定，取哪个参数是根据executor有没有缓存来决定的(**也就是是否有运行的rdd调用了cache或者persist**)。空闲时间超过指定的时间就会被定时任务移除。

其中`spark.dynamicAllocation.executorIdleTimeout`的默认值是30s，`spark.dynamicAllocation.cachedExecutorIdleTimeout`的默认值却是无穷大，因此动态资源任务要额外注意spark.dynamicAllocation.cachedExecutorIdleTimeout的设置。

当一个task开始时，会将运行的executor标为忙碌，然后task结束后，将对应的executor标记为空闲。一旦executor被标记为空闲，就会开始计时，直到达到超时时间然后被移除。

> **在Spark-2.1.0版本中有个bug**，当remove掉一个Executor时，由于numExecutorsTarget值没有更新，如果此时numExecutorsTarget的值刚好是maxNumExecutors，就会导致明明job需要新的executor，但是就是申请不到的问题（numExecutorsTarget >= maxNumExecutors条件始终成立）。

> 相关issue：[https://issues.apache.org/jira/browse/SPARK-21834][https_issues.apache.org_jira_browse_SPARK-21834]

## 三、问题定位 ##

由于集群的spark版本刚好是2.1.0，根据issue [https://issues.apache.org/jira/browse/SPARK-21834][https_issues.apache.org_jira_browse_SPARK-21834] ，确实会导致executor remove后无法申请到足够的executor数量的问题。（具体参考上面一节的Executor Remove小节）

上面的issue虽然解释了为什么最后只剩下一个executor在运行，又不申请新的executor的问题。但是还有一个问题还没解决：为什么在待运行的task还有很多的情况下，会有那么多的executor空闲时间超过30s？

仔细看了下spark调度task的相关源码，发现这个问题是由于 Task的本地化调度导致的。

### Spark Task的本地化调度 ###

先简单说下Task的调度。task调度的源头从CoarseGrainedSchedulerBackend类开始。CoarseGrainedSchedulerBackend类在启动时，会启动一个定时任务线程"driver-revive-thread"，每隔一定时间运行一次，间隔由参数spark.scheduler.revive.interval时间运行一次，默认是1s。

简单来讲，这个定时任务就是收集一下存活的executor，然后最终调用TaskSchedulerImpl\#resourceOffers(executors)方法来处理。然后由TaskSchedulerImpl根据调度算法选择需要调度的TaskSetManager列表(**排好序的，有先后顺序，这个和参数spark.scheduler.mode配置的调度算法有关**)，之后逐个遍历这些TaskSetManager进行调度。另外，每个task运行结束后也会再次调用TaskSchedulerImpl\#resourceOffers(executors)方法检查是否要继续调度下一个task。

TaskSetManager是由多个task组成的，代表一个stage的所有task。下面是TaskSetManager的一些关键参数:

//PROCESS_LOCAL 级别对应的task列表，key是executorId，value是task列表
    private val pendingTasksForExecutor = new HashMap[String, ArrayBuffer[Int]]
    //NODE_LOCAL 级别对应的task列表，key是host，value是task列表
    private val pendingTasksForHost = new HashMap[String, ArrayBuffer[Int]]
    //RACK_LOCAL 级别对应的task列表，key是rack，value是task列表
    private val pendingTasksForRack = new HashMap[String, ArrayBuffer[Int]]
    //不关心级别的task列表
    var pendingTasksWithNoPrefs = new ArrayBuffer[Int]
    //所有待执行的task
    val allPendingTasks = new ArrayBuffer[Int]
    //根据Task的情况获取当前TaskSet对应的几种级别。正常情况下依次是 PROCESS_LOCAL NODE_LOCAL NO_PREF RACK_LOCAL ANY
    var myLocalityLevels = List[TaskLocality]
    //对应级别的最长等待时间
    var localityWaits
    //最后一个task的启动时间
    var lastLaunchTime

本地化级别从低到高依次是 PROCESS\_LOCAL、NODE\_LOCAL、NO\_PREF、RACK\_LOCAL、ANY。高级别的task列表都会包含低级别的task列表。也就是说pendingTasksForExecutor中有的task，在pendingTasksForHost中也一定会存在。对于pendingTasksForHost和pendingTasksForRack也是同理。

由于TaskSchedulerImpl调度具体TaskSetManager的代码比较绕，为了方便理解，这里还是直接上伪代码：

//遍历当前TaskSetManager中存在的所有本地化级别，一般依次是 PROCESS_LOCAL NODE_LOCAL NO_PREF RACK_LOCAL ANY
    //遍历的目的主要是为了优先调度低级别的task（如果TaskSetManager的本地化级别已经推进到高级别，此时就可以返回来调度低级别的task）
    for(currentMaxLocality <- taskSet.myLocalityLevels){
        //遍历所有的executor
        for(executor <- executors){
            //调用TaskSetManager的方法获取本次要调度的task列表
            taskDescriptionList = taskSet.resourceOffer(executor.id,execute.host,currentMaxLocality){
                //spark的黑名单机制，如果executor的id或者host在黑名单里面，就过滤
                if(inblack(executor)){
                    return null;
                }
                 
                allowedLocality = maxLocality
     
                if(maxLocality != TaskLocality.NO_PREF){
                    allowedLocality = getAllowedLocalityLevel(curTime){
                        while(currentLocalityIndex < myLocalityLevels.length - 1){
                            //获取当前TaskSetManager所处的级别
                            currentLocality = myLocalityLevels(currentLocalityIndex)
                            //判断该级别是否有任务需要运行
                            moreTasks = moreTasksToRunIn(currentLocality)
                            //如果当前级别没有任务需要运行了，就推进一个级别
                            if(!moreTasks){
                                currentLocalityIndex += 1
                            //判断当前级别的延迟时间是否超过了指定的值
                            }else if(curTime - lastLaunchTime >= localityWaits(currentLocalityIndex)){
                                lastLaunchTime += localityWaits(currentLocalityIndex)
                                currentLocalityIndex += 1
                            }else{
                                return myLocalityLevels(currentLocalityIndex)
                            }
                        }
                        return myLocalityLevels(currentLocalityIndex)
                    }
                    //尽量调度低级别的task
                    allowedLocality = min(allowedLocality , maxLocality)
                }
                //根据allowedLocality级别获取对应的task
                return dequeueTask(execId, host, allowedLocality).map{...}
            }
            //taskDescriptionList 不为空则调度这些task（这里已经知道了哪个task要发给哪个executor调度了）
        }
    }

从代码可以看出，spark的本地化调度是一个逐级推进的过程。从最低级别的PROCESS\_LOCAL开始。推进到下一级有两个条件：

1.  该级别的已经没有task需要运行
2.  距离上一次该级别的最后一个task运行时间已经超过了spark.locality.wait的时间，这个值默认是3s

**如果没有及时推进到下一级，TaskSetManager会不断尝试根据当前的本地性级别取task，如果有一些task已经无法满足该级别了，即使executor空闲再多也不会把这些task分配过去。因此才导致了大量的executor空闲了超过30s的时间。**

### 本地化调度级别没有及时推进导致的问题 ###

结合本地化调度的推进过程又去看了下ApplicationMaster的日志，发现该任务task的运行时间很快，基本都是1秒内完成。**并且在NODE\_LOCAL这个级别，在许多executor已经没有task可以执行的情况下，有某几个executor还一直在执行**。由于没有推进到下一级，那些没有task运行的executor就会空闲在那，最终空闲时间超过30s后被ExecutorAllocationManager给kill了。后面级别也推进到了RACK\_LOCAL级别，但是由于[https://issues.apache.org/jira/browse/SPARK-21834][https_issues.apache.org_jira_browse_SPARK-21834] 这个issue，executor也没有恢复回来，一直保持在一个。

这个问题的形成主要有两个：

1.  task运行过快，低于spark.locality.wait的值
2.  在\*\*NODE\_LOCAL级别上，\*\*某几个executor长时间有task运行，其他大量的executor又没有task运行。可能是刚好如此分布，也可能是datanode数据倾斜造成的

## 四、解决方案 ##

### 1、问题一 ###

executor因为空闲被remove时，还有很多待运行的task。为什么executor会空闲长达30s不接受任务运行呢？

1.  调低spark.locality.wait的值，让本地化级别可以较快的推进到下一级。但是这样也会减弱spark的数据本地化机制，导致任务运行变慢。
2.  如果是datanode的数据倾斜导致的，可以对hdfs集群进行一次balance。
3.  如果修复了问题二的话，其实这个问题只会导致任务运行时间稍微变慢一点，并不会对任务有太大的影响。

### 2、问题二 ###

executor只剩下一个的时候，待运行的task数量也还有很多。那为什么没按照动态资源的规则继续申请新的executor呢？

根据[https://issues.apache.org/jira/browse/SPARK-21834][https_issues.apache.org_jira_browse_SPARK-21834]的patch即可解决

### 3、总结 ###

其实问题一严格来说也不能算bug，一定程度上只能说是策略的问题。如果解决了问题二，问题一最多导致任务的运行效率小幅度降低。因此，如果解决了问题二，对于问题一我们可以采用参数调优的方式来优化任务的性能。

[https_issues.apache.org_jira_browse_SPARK-21834]: https://issues.apache.org/jira/browse/SPARK-21834