Spark报FetchFailedException（远程Executor已死）如何解决？

报错信息 #

Caused by: org.apache.spark.SparkException: Job aborted due to stage failure: ResultStage 9 (runJob at FileFormatWriter.scala:237) has failed the maximum allowable number of times: 4. Most recent failure reason: org.apache.spark.shuffle.FetchFailedException: The relative remote executor(Id: 156), which maintains the block data to fetch is dead. 
at org.apache.spark.storage.ShuffleBlockFetcherIterator.throwFetchFailedException(ShuffleBlockFetcherIterator.scala:747) 
at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:662) 
at org.apache.spark.storage.ShuffleBlockFetcherIterator.next(ShuffleBlockFetcherIterator.scala:70) 
at org.apache.spark.util.CompletionIterator.next(CompletionIterator.scala:29) 
at scala.collection.Iterator$$anon$11.next(Iterator.scala:410) 
at scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:435) 
at scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:441) 
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409) 
at org.apache.spark.util.CompletionIterator.hasNext(CompletionIterator.scala:31) 

原因分析 #

核心问题是持有Shuffle数据块的远程Executor（ID:156）已死亡，导致Reducer无法拉取所需的Shuffle数据，最终作业因Stage重试次数耗尽而失败。Executor死亡的常见诱因包括内存不足（OOM）、磁盘IO过载、节点硬件故障、网络失联、资源管理器（如YARN）强制回收资源等。

解决办法 #

1. 定位Executor死亡的根本原因 #

• 查看Executor所在节点的系统日志（如/var/log/messages），排查是否存在内存耗尽、磁盘满、网络中断等系统级问题。
• 登录Spark UI的Executors页面，找到ID为156的Executor，查看其完整日志，定位具体崩溃原因（比如OOM报错、磁盘写入失败等）。
• 若为YARN集群，通过yarn logs -applicationId <app-id>查看对应容器的日志，获取更详细的错误信息。

2. 调整Spark资源配置 #

• 解决OOM问题：调大spark.executor.memory（比如从4G改为8G），同时降低spark.executor.cores（比如从4改为2），减少单Executor的任务并发数，降低内存压力。
• 优化内存分配：开启堆外内存，设置spark.memory.offHeap.enabled=true和spark.memory.offHeap.size=4g，利用堆外内存缓解堆内内存不足的问题。
• 增加Driver内存：调大spark.driver.memory，避免Driver端内存不足影响作业调度和Shuffle元数据管理。

3. 优化Shuffle相关配置 #

• 启用外部Shuffle服务：设置spark.shuffle.service.enabled=true，Shuffle数据会由独立的Shuffle服务进程托管，即使Executor死亡，Reducer依然可以从Shuffle服务拉取数据，从根源避免这类FetchFailed问题。
• 调整Shuffle IO参数：
  • 调大spark.shuffle.file.buffer（比如改为64k），减少Shuffle文件的磁盘IO次数。
  • 降低spark.reducer.maxSizeInFlight（比如改为32m），控制Reducer每次拉取的Shuffle数据量，避免内存溢出。
• 跳过不必要的排序：当Shuffle分区数较少时，设置spark.shuffle.sort.bypassMergeThreshold=500，跳过排序阶段，减少计算开销。

4. 优化作业代码 #

• 减少Shuffle数据量：替换groupByKey为reduceByKey或aggregateByKey，这类操作会在Map端做预聚合，大幅减少Shuffle阶段的数据传输量。
• 过滤无效数据：在作业早期过滤掉不需要的行或列，减少整体数据处理规模。
• 解决数据倾斜：检查是否存在数据倾斜问题（比如某几个Key对应的数据量远大于其他Key），可以通过加盐、拆分分区等方式缓解，避免单个Task占用过多内存导致Executor崩溃。
• 调整分区数：合理设置spark.sql.shuffle.partitions（默认200），确保Shuffle分区数与数据量匹配，避免分区过多导致内存开销大，或分区过少导致任务过载。

5. 优化集群环境 #

• 检查磁盘空间：清理Executor节点的磁盘垃圾，确保有足够的空间存储Shuffle文件和作业临时数据（至少保留10%以上的空闲空间）。
• 优化网络：检查集群节点间的网络带宽和稳定性，避免因网络延迟或丢包导致Executor与Driver失联。
• 调整资源管理器配置：若为YARN集群，调整yarn.nodemanager.resource.memory-mb和yarn.scheduler.maximum-allocation-mb等参数，确保资源分配足够，避免因资源抢占导致Executor被强制kill。

内容的提问来源于stack exchange，提问作者湘晗刚