executor没有被初始化（Spark MetadataFetchFailedException 问题排查）

本文目录

• Spark MetadataFetchFailedException 问题排查
• ThreadPoolExecutor线程池

Spark MetadataFetchFailedException 问题排查

***隐藏网址***

业务反馈某个任务运行失败，看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异常退出的最大可能就是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的日志，都没找到相关的出错日志。

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

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

如果某个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。

后面无意看到一个成功的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的大小。因此我们有两个方向调整任务的参数：

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

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

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

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了。

解决办法：

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

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

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

解决方法：

ThreadPoolExecutor线程池

当我们需要实现并发、异步等操作时，可以使用ThreadPoolExecutor。

ThreadPoolExecutor

线程池：

系统中，我们创建（extend Thread/implement Runnable）、销毁（正常run方法完成后线程终止）线程的代价是比较高昂的。
如果频繁地创建和销毁进程，会大大降低系统运行效率和吞吐量。
线程池使得线程可以被复用，避免了线程频繁创建和销毁的开销，提高系氏芦橘统的运行效率和吞吐量。

实例

ThreadPoolExecutor.execute(new Runnable () {});

相关概念：

Task任务：new Runnable () {}
任务就是一个Runnable的对象，任务的执行方法就是该对象的run方法。

缓冲队列：workQueue

一个阻塞队列。

BlockingQueue《Runnable》 workQueue;

corePoolSize：核心线程数
核心线程会一直存活，即使没有任务需要执行。
当线程数小于核心线程数时（还未满，就会一直增），即使有线程空闲，线程池也会优先创建新线程处理。

maxPoolSize：最大线程数
当线程数大于corePoolSize，且任务队列已满时。线程池会创建新线程来处理任务，直到线程数量达到maxPoolSize。

execute(Runnable)

通过execute将一个任务交由线程池管理。

当一个任务通过execute方法欲添加到线程池时，线程池采用的策略如下（即添加任务的策略）：

1、如果此时线程池中的数量小于corePoolSize，即使线程池中的线程歼团都处于空闲状态，也要创建新的线程来处理被添加的任务哗升。

2、如果此时线程池中的数量等于corePoolSize，但是缓冲队列workQueue未满，那么任务被放入缓冲队列。

3、如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量小于maximumPoolSize，建新的线程来处理被添加的任务。

4、如果此时线程池中的数量大于corePoolSize，缓冲队列workQueue满，并且线程池中的数量等于maximumPoolSize，那么通过handler所指定的策略来处理此任务。

如下图：

希望对您有所帮助！~