一、问题描述

业务反馈某个任务运行失败，看Spark History Server发现多个Stage报如下错误：

从UI以及异常的信息来看，是下游Stage做Shuffle Read时发现数据丢失，之后上游Stage和本Stage都重新调度计算，但是重试又发生类似的情况，直到重试3次后整个application失败。

后面application又自动进行重试，但是依旧失败了，因此判断这不是偶然的情况。

所以问题只有一个：为什么下游Stage要获取数据时，找不到上游Stage输出的数据了？

二、问题定位

shuffle的过程主要是上游Stage将数据写到磁盘，然后下游Stage通过Executor的BlockManager来拉取数据。如果下游Stage要拉取数据时Executor已经异常下线了，就会导致下游Stage拉取不到数据。这时候就会报MetadataFetchFailedException。

看了下Driver的日志，确实发现很多 Lost executor 日志（如果Executor不是Driver主动通知退出的，Driver发现Executor长时间没有响应就会输出这条日志）。

随便找了台Lost Executor的日志，确实有些问题：

这是Executor上最后的几十行日志。一般task开始和结束都会输出日志，但是在这个日志中，我们就看到某几个Task开始运行的日志，之后日志就断了。说明Executor在这个时候异常退出了。

Executor异常退出的原因猜测

1、OOM导致Executor异常退出

Executor异常退出的最大可能就是OOM了。这个任务配置的Executor内存是10g，core数量是8个，一个task需要的cpu是1个，因此task并行度是8。如果task需要的内存大于 10/8=1.25g的话，那很可能会导致OOM。

但是观察了下spark History server上数据量的大小，发现各个stage处理的数据量都不大，最多几十G，stage的task数量一般都是1000个，分摊下来一个task也就处理几百M那里，不至于导致OOM。当然，不排除数据倾斜的问题。

最重要的是，如果Executor发生OOM，是会有出错日志的，但是看了几个Executor的日志，都没找到相关的出错日志。

2、linux OOMKiller

还有一种情况是linux在系统内存使用紧张时会根据一些算法kill某些进程，Executor可能就会被kill掉。

后面看了下那个时间点机器的使用内存，发现还有大量的剩余内存。因此和OOMKiller无关

3、因磁盘问题Executor被yarn Kill

如果某个NodeManager在运行过程中工作磁盘使用率达到了一定的阀值，该NodeManager会被标记为UnHealth，同时在上面运行的Container都会被停止。

用df -h命令检查了下机器的磁盘使用情况，也没什么问题。因此排除这种可能

和判断NodeManager是否健康的配置有：

yarn.nodemanager.disk-health-checker.enable: 是否开启磁盘健康检查，默认是true，表示开启

yarn.nodemanager.disk-health-checker.min-healthy-disks：NodeManager上最少保证健康磁盘比例，当健康磁盘比例低于该值时，NodeManager不会再接收和启动新的Container，默认值是0.25，表示25%；

yarn.nodemanager.disk-health-checker.max-disk-utilization-per-disk-percentage：一块磁盘的最高使用率，当一块磁盘的使用率超过该值时，则认为该盘为坏盘，不再使用该盘，默认是90，表示90%，可以适当调低；

yarn.nodemanager.disk-health-checker.min-free-space-per-disk-mb：一块磁盘最少保证剩余空间大小，当某块磁盘剩余空间低于该值时，将不再使用该盘，默认是0，表示0MB。

4、因内存问题Executor被yarn Kill

后面无意看到一个成功的Stage的某个失败Task日志，才知道yarn会因为内存问题会kill Executor：

异常信息也很明了，Executor使用的内存超过了限制，因此被Yarn Kill掉。

spark.yarn.executor.memoryOverHead默认配置是ExecutorMemory的10%，也就是1G。所以Executor在yarn这边申请的内存是11G。但是spark.yarn.executor.memoryOverHeap是堆外内存，主要用于JVM自身，字符串, NIO Buffer等开销，因此没做限制。

如果运行的task数量过多，OverHead的使用就会超过1G，最终Executor总的使用内存超过11G，yarn有个container内存使用检测机制，如果发现有container内存使用超标，就会主动kill这个Container。

问题总结

这个问题的主要原因就是任务的task并行度是8，但是OverHead大小只有1G，在任务运行过程中不时的OverHead区域超过了1G，最终导致Executor被yarn给kill了。

举个例子，比如某个上游Stage的task运行成功将数据写入Executor A后，下游Stage的task又分发到这个Executor A上，这时运行过程中Executor A的OverHead区域使用超过了1G，整个Executor 被kill，这样Executor A上的shuffle数据就丢失了。后面的重试又不断的重复这样的场景，直到整个任务失败。

三、解决方案

通过调整任务的参数可以解决这个问题。这个问题的主要原因在于OverHead的大小。因此我们有两个方向调整任务的参数：

1. 通过spark.yarn.executor.memoryOverHead调大OverHead的大小
2. OverHead使用太多主要还是task并行度的原因，也可以调小task的并行度来降低overHead的使用

四、扩展：Executor因内存问题被Yarn Kill的情况

Executor运行时会告诉yarn要申请的资源数量，如果要申请的内存超过yarn配置的 yarn.scheduler.maximux-allocation-mb值，yarn就会拒绝Executor的启动。

Executor启动后，yarn也有 Container 的内存监控机制。如果运行过程中Executor实际使用的内存超过了申请的内存，yarn发现了就会主动kill 这个Executor。比如Executor申请资源时指定要10G，但是运行过程过实际使用超过了10G，那么yarn就会主动kill这个Executor。

有两种情况可能导致Executor实际使用的内存超过预期值：

1、Overhead 区域使用超过预期值

overhead的大小由spark.yarn.executor.memoryOverHead来配置，如果没配置，默认是ExecutorMemory的10%。

Executor向yarn申请内存时会自动算上overhead，但是还是会有overhead大小不够用的情况。

这种一般都发生在Executor同时运行的task数量比较多的情况，overhead区域主要用于JVM自身，字符串, NIO Buffer（Driect Buffer）等开销，如果task并发度太高，就会导致overhead区域不够用的情况。

因为overhead是堆外内存，因此它不会受JVM内存限制。但是overhead使用超过了预期的值后，yarn就开始kill这个 Executor了。

解决办法：

• 调大overhead的值
• 降低task的并行度，task并行度有 spark.executor.cores / spark.task.cpus 决定

2、Executor又开启了子进程导致总内存使用超出预期

yarn对container内存的判断不单单只判断Executor的内存使用量，如果Executor又开启了一个子进程，这个子进程使用的内存也会算在Container的内存使用里面。

也就是说，yarn对container内存判断算的是整个Executor进程树的总内存大小。

典型的场景就是Executor中又进行了shell调用，shell运行的子程序又占用了一定大小，最终导致超过了预期值，然后Executor被yarn kill。

解决方法：

• 保证开启的子进程的内存使用，同时设置好合适的ExecutorMemory