Spark作业失败排查指南：Executor丢失、数据倾斜与Shuffle调优实践

1. 一份报错日志的阅读顺序：先分清作业死在哪一层

先说个真实场景。凌晨两点，某开发者的批处理作业跑了近六个小时，眼看就要到收尾阶段，突然一堆红色报错刷屏，Stage 12 反复重试了四五次，最后整个 Application 被标记为 Failed。他当时的第一反应是翻 Executor 日志，结果翻了半天，除了几行Container exited with a non-zero exit code之外什么也没看出来。这不是个例，我遇到过太多人把时间浪费在错误信息最表面的一层。

Spark 作业的失败原因虽然五花八门，但基本可以按"层"来划分：提交层（Client 提交参数错误、依赖缺失）、调度层（资源不足、队列拒绝）、执行层（Executor 崩溃、任务重试）、数据层（倾斜、序列化失败）、存储层（HDFS 写入失败、目标表冲突）。拿到一个失败任务，第一步不是去看堆栈，而是先确定它死在哪一层。判断依据很简单：看日志是从哪个进程打出来的，以及失败发生在哪个 Stage。

1.1 打开 Spark UI 之后的第一个动作

不管报错多吓人，先登录 Spark UI，依次看三个页面：Jobs 页面看整体执行到了哪个 Job；Stages 页面看具体是哪个 Stage 失败、失败的任务有多少、Shuffle Read/Write 量是多少；Executors 页面看有没有 Executor 处于 Dead 状态。

大多数情况下，问题在这个环节就能暴露出一半真相。比如 Executors 列表里出现一个灰掉的节点，任务失败原因列着ExecutorLostFailure，说明执行进程被强制杀掉了；如果 Stage 里显示某个 Task 反复重试，失败原因写的是java.lang.OutOfMemoryError，那就是执行端内存不够。UI 上的信息永远是排查的第一现场，比你去登服务器翻 user logs 快得多。

1.2 日志里常见的几类"暗号"

先说几个我几乎每周都会见到的典型错误串，以及它们最常对应的真实原因：

这些信息往往只是"引子"，真正的根因需要继续往下挖。但至少它们能告诉你，该往哪个方向去查。

1.3 Driver 端失败与 Executor 端失败：两条完全不同的排查线

这是很多人一开始没搞清的关键点。

如果失败发生在Driver 端，整个 Application 会直接退出，日志里能看到你的main函数堆栈，伴随的通常是Driver stacktrace。常见原因包括：Driver 内存塞不下collect()回来的数据、外部依赖在 Driver 上不存在、提交参数里配置了错误的主类。排查时主要看 Driver 的 stdout/stderr，以及提交命令本身。

如果失败发生在Executor 端，Application 本身不一定立刻退出，而是表现为某个 Stage 的 Task 反复失败，重试到一定次数后 Spark 才把整个 Job 标记为失败。这时真正有价值的日志在 Executor 的 stdout/stderr 或者 YARN 的 Container 日志里。很多人在这一步走错方向——盯着 Driver 日志找 Executor 报错，自然什么都找不到。

2. Executor 丢失与内存型故障：别把锅全甩给集群

ExecutorLostFailure是 Spark 作业失败里出现频率最高的一类，而它背后九成是内存问题。

2.1 表象与真实根因的错位

一次典型的报错长这样：

org.apache.spark.SparkException: Job aborted due to stage failure: Task 37 in stage 12.0 failed 4 times, most recent failure: Lost task 37.1 in stage 12.0 (TID 9821, host-xx-xx): ExecutorLostFailure (executor 5 exited caused by one of the running tasks) Reason: Container killed by YARN for exceeding memory limits.

注意最后一句：Container killed by YARN for exceeding memory limits。这说明 Executor 进程实际占用的物理内存超过了 YARN 给它分配的容器上限。很多人以为这是"集群内存不够"，但其实更常见的是你自己给 Executor 堆内内存配置不合理，或者堆外内存被漏算了。

Executor 进程占用的内存 = 堆内（Spark Heap）+ 堆外（Overhead）+ 其他（如 JVM 自身、线程栈、Direct Memory、Shuffle 的 Netty Buffer 等）。spark.executor.memory只配置了堆内部分，堆外默认是spark.executor.memoryOverhead = max(executorMemory * 0.1, 384m)。如果你把堆内配到 8G，同时并行度又很高，线程数暴涨，堆外很容易超掉那个 10% 的默认值。这时候 YARN 会直接杀死容器，而你的堆内可能根本没到上限。

2.2 Driver OOM 的元凶：collect() 与结果集回溯

另一类高频内存故障是 Driver OOM。典型代码长这样：

# 错误示范：把小表数据全部拉回 Driver result = df.groupBy("region").count().collect() # 如果分组特别多，或者 DataFrame 字段很多，结果集可能瞬间把 Driver 内存打满

再强调一种隐蔽情况：df.count()在 Spark 里会触发一个 Job，如果前面的宽依赖链特别长，Driver 需要保存执行计划的中间状态；df.collect()更是直接把整份数据拉到 Driver 堆内。一旦数据量在千万级以上，Driver 的java.lang.OutOfMemoryError: Java heap space几乎必现。

我习惯的做法是：凡是最终结果超过几十万行的，一律不 collect，直接写回表或文件；需要预览数据时用df.limit(1000).collect()配合 DataFrame 的采样方法，而不是全量拉回。

2.3 一套可复用的内存参数基线

下面这套参数是我在多个集群上调过之后沉淀下来的起点（注意是起点，不是终点），单位均为 GiB/核：

spark.executor.memory=4G spark.executor.memoryOverhead=2G spark.driver.memory=2G spark.driver.memoryOverhead=1G spark.executor.cores=4 spark.memory.offHeap.enabled=false

关键点：堆外内存宁可多给一点，尤其在 Shuffle 数据量大的场景下，Overhead 配 2G 比默认值稳很多。另外，spark.executor.cores不建议超过 5，因为 JVM 线程数量越多，堆外开销越大，GC 表现也会变差。

如果还出现 GC 频繁导致的任务停顿，可以打开 G1：

spark.executor.extraJavaOptions=-XX:+UseG1GC -XX:MaxGCPauseMillis=200

配合 Spark UI 的GC Time列来判断：如果某个 Executor 的 GC 时间占了任务运行时间的 30% 以上，说明内存效率有问题，这时候不是单纯加内存就完事，还要看是否做了过多无效的 Shuffle 或缓存。

3. 数据倾斜：作业不报错但永远跑不完的真凶

和内存溢出相比，数据倾斜更隐蔽——它不报错，你的 Job 也在持续推进，但就是有一个 Task 卡在 99%，整个 Stage 等你几个小时。我排过的大量"作业跑到一半卡死"案例，最终根因都是倾斜。

3.1 怎么用 Spark UI 确认倾斜

看两个指标：Stage 里各个 Task 的 Shuffle Read Size 和 Duration。正常情况各 Task 的耗时是接近的钟形分布；倾斜时你会看到某个 Task 的 Shuffle Read 达到几个 GB，而其他 Task 只有几十 MB，或者单个 Task 运行时间是平均值的好几十倍。截图都不用贴，光看这两个数就能判断。

另一个指标是Input Size / Records：如果某个分区里的记录数比其他分区高一个数量级以上，那就是典型的倾斜特征。很多情况下，这个 Task 还会伴随 GC 时间异常，因为它要处理的数据量太大。

3.2 倾斜最常见的三类场景

第一类：Join 倾斜。事实表关联维表时，某个热门 key（比如头部用户的 ID）把大量数据打进同一个分区。表现为 Stage 中间某个 Task 的 Shuffle Read 巨大。

第二类：GroupBy 倾斜。对某个高基数维度做聚合，但极少数 key 占了总数据量的绝大部分——最典型的就是"省份""渠道"这类分布极不均匀的字段。

第三类：distinct/countDistinct 倾斜。很多新手不知道 countDistinct 会触发两次 shuffle，第一次是去重前的 map-side combine，第二次是按 key 分组。倾斜发生在第一轮时，某些 key 会拖垮整个阶段。

3.3 我实践过的几种解法

加盐。对倾斜 key 打散后聚合，再合并结果。以 groupBy 为例：

from pyspark.sql.functions import col, concat, lit, rand, split # 给倾斜字段加随机后缀，把一个大分区拆成 N 个分区 df_skewed = df.withColumn("salt", concat(col("key"), lit("_"), (rand() * 10).cast("int"))) # 先按加盐后的 key 聚合 agg1 = df_skewed.groupBy("salt", "other_cols").agg(...) # 再按原始字段汇总（去掉盐后缀） agg2 = agg1.withColumn("key", split(col("salt"), "_")[0]).groupBy("key", "other_cols").agg(...)

注意：这个方案里盐的数量（代码里的 10）不是越大越好，太大反而增加 Shuffle 量和中间聚合开销；合理值一般取 5~20，取决于数据量。

倾斜 Join 用 Map-Side Broadcast。如果倾斜发生在事实表和维表 join，而维表不超过 1GB 左右，直接强制广播：

from pyspark.sql import functions as F joined = fact.join(F.broadcast(dim), on="key", how="inner")

广播后小表不会参与 shuffle，事实表按自身分区直接和广播表本地关联，倾斜自然缓解。这也是我处理 join 倾斜时最先尝试的方案，因为它改动最小、收益最直接。

对倾斜 key 单独过滤处理。如果只是少数几个 key 导致倾斜，可以先用侧输出（side output）把倾斜 key 的数据拆出来单独走一套逻辑，正常数据走原逻辑，最后 union 回去。这种方案定位精确、不容易引入副作用，缺点是代码复杂性上来了，需要额外维护一份"倾斜 key 清单"。

提示：排查倾斜前先确认你集群有没有开启动态资源分配或 spark.sql.shuffle.partitions 是否明显偏小。默认 200 个分区对于几十亿行数据来说往往不够，有时候把分区数调大到 1000 也能缓解倾斜——因为倾斜的"大头"被拆碎了。

4. Shuffle 与序列化故障：藏在中间环节的坑

内存问题解决之后，第二个高频"翻车点"是 Shuffle 和序列化。这两类问题往往隐藏得很深，因为它们的报错信息极具迷惑性。

4.1 FetchFailedException：典型但不是每次都要重算

org.apache.spark.shuffle.FetchFailedException: Failed to connect to host-xx:port

这个报错的具体成因有三个：Executors 在 Shuffle 过程中被杀死导致上游数据块丢失；节点磁盘故障导致 Shuffle 中间文件读取失败；分区数过大导致每个 Shuffle 文件块太小、文件句柄过多。大部分情况下，Spark 会自动重试 Fetch 失败的任务，但如果是 Executor 反复丢失，每次 Fetch 重试都会带上更多的上游节点，最终可能整个 Stage 全部失败。

我的常规做法是：先确认是不是伴随 Executor 丢失（回到第 2 节），如果是，先解决内存问题；如果不是，检查 Executor 节点的磁盘可用空间和 IO 异常。集群里如果大量出现FetchFailed，还需要考虑要不要开spark.shuffle.service.enabled——有了外部 Shuffle Service，即使 Executor 被回收，shuffle 数据依然由专门的常驻服务持有，能显著减少这类失败。

4.2 Task not serializable：闭包污染的经典案例

org.apache.spark.SparkException: Task not serializable

这个报错几乎所有写过 Spark 的人都会遇到。根因一句话就说清：你在 map/filter/foreach 这类算子里引用了外部对象，而该对象不可序列化。

最常见的坑是：你定义了一个普通 class 的对象，直接用在闭包里，而这个 class 没有实现Serializable。更隐蔽的是匿名内部类、单例里持有连接对象（比如数据库连接池）或者 lambda 捕获了不可序列化的上下文。

// 错误示范：自定义类未序列化 public class IdGenerator { public String gen(int x) { return "id-" + x; } } // rdd.map(r -> new IdGenerator().gen(r)) // 如果每次都在算子内部 new，问题不大 // rdd.map(r -> generator.gen(r)) // 如果引用了外部 generator，就必须可序列化

修复思路就三类：类实现Serializable；把不可序列化的对象放在算子内部创建；或者用rdd.mapPartitions在分区级别创建一次连接对象。第三种方式不仅解决序列化问题，还能减少连接创建次数，提升性能。

4.3 Kryo 序列化用不用？我的结论

很多人把 Kryo 当成"性能优化"来开，但我的经验是：Kryo 对 RDD 序列化性能的提升是实打实的，但对 Spark SQL 的影响没那么明显，因为 Spark SQL 的 InternalRow 用的是独立编码。如果你的任务以 RDD API 为主、Shuffle 量又大，开 Kryo 效果明显；如果全是 DataFrame API，收益有限。

开了 Kryo 后有一件事必须做：把所有自定义类注册到spark.kryo.registrator里，否则 Kryo 可能抛ClassNotFoundException或序列化异常。另一个要注意的是 Kryo 对某些复杂嵌套结构支持不是特别好，遇到反序列化报错时，可以先回退到 Java 序列化做对比定位。

4.4 写入阶段的小文件与分区提交问题

作业跑完最后一步写入时失败，也是排查热点。常见报错包括写入时目标分区已存在、PartitionAlreadyExistsException，以及因为写太多小文件导致 HDFS 目录操作超时。

这里有个很反直觉的点：Spark 写入 HDFS 时是按 Task 粒度落盘的，一个 Task 一个文件。你的 DataFrame 有 2000 个分区，就会写 2000 个小文件；如果还要 dynamic partition 插入，小文件问题会指数级放大。遇到写入阶段超时或 Namenode 压力大，先检查最终输出目录的文件数与文件大小是否离谱。常规优化是写入前repartition或coalesce控制分区数，再开spark.sql.adaptive.coalescePartitions.enabled=true让 Spark 自动合并小分区。

5. 一套可落地的排查流程：从报错到定位的七个步骤

说了这么多具体故障，最后沉淀一套我在团队里推行的排查流程。它的核心原则是：先看 UI，再定层次，后改参数，最后看代码。很多人一上来就翻代码，我不同意——你得先知道问题是不是出在资源层，再来讨论逻辑。

5.1 七步排查法

1. 锁定失败的 Job 与 Stage：在 Spark UI 的 Jobs 页面找到失败的 Job，记下失败的 Stage 编号。
2. 看 Stage 详情：记录 Task 总数、成功/失败数、最大耗时 Task、Shuffle Read/Write、GC 时间。
3. 看 Executors 状态：检查是否有 Executor 處於 Dead 或异常退出，记录退出时间与当时正在跑的任务。
4. 区分 Driver 与 Executor 日志：去对应进程的日志里找真正的堆栈信息，别混着看。
5. 按失败类型套模板：内存问题看第 2 节，倾斜看第 3 节，Shuffle/序列化看第 4 节。
6. 复现与最小化：用相同数据量的子集单独跑一遍，能复现就缩小范围；不能复现则要考虑是否偶发资源问题。
7. 改配置或改代码后做 A/B 验证：不要同时改两个变量，保持单一变量原则。

5.2 一个真实案例的完整复盘

我这里有一个典型的"运行三小时、最后阶段失败"的案例。某公司一个批处理作业每天晚上跑，处理的是一张几十亿行的行为日志表，流程是：读源表 → 清洗 → 按用户维度聚合 → join 用户维表 → 结果写回结果表。

第一晚失败时的报错：Stage 47 任务失败，原因FetchFailedException，重复重试后整个 Application 失败。

我当时按流程走：UI 里看 Stage 47 的 Shuffle Read 发现有四个 Task 的量明显异常——一个 Task 拉到 3.2GB，其他都在 50MB 左右。这立刻指向数据倾斜，而不是网络或磁盘问题。进一步看聚合阶段的 key 分布，发现某个头部用户 ID 的日志量占全天的 0.3%，但在按用户分桶时被分到了同一个分区，导致该分区数据量是其他分区的百倍。

修复方案上，我们当时因为临时没空改代码，先做了两个低风险操作：把spark.sql.shuffle.partitions从默认 200 调到 1200，同时给该聚合阶段开启spark.sql.adaptive.skewJoin.enabled=true。第二天跑完，运行时间从 6 小时压到 3 小时。后续再在代码层面对该 key 做了加盐拆分，彻底解决。

这个案例的启发是：兜底参数能救急，但根治还是要理解数据分布。参数改完效果立竿见影时，很多人就停了，但数据持续增长后问题可能复发。根源的加盐或广播方案才是真正该做的。

5.3 该加日志的地方一定要加

排查过程中最怕"黑盒"。建议在所有需要 Debug 的环节主动输出日志，比如每个 Shuffle 之后记录df.count()级别或分区数变化，写入关键步骤时打印出分区数、记录数、预估大小。一个简单例子：

logger.info(f"Stage A after groupBy: partitions={df.rdd.getNumPartitions()}, count={df.count()}")

记录这些，不只是为了排查当下问题，更是给后续接手的人留下线索。很多 Spark 作业失败，真相埋藏在数据量变化的链条里——中间结果集突然膨胀，到最后一个环节才爆掉，没有中间日志你永远只能猜。

6. 写在最后：几条吃了亏才学会的排查习惯

最后分享几个我在实际运维中形成的习惯，谈不上标准答案，但都是实打实踩过坑换来的。

第一，Spark UI 的日志要保留至少一周。很多作业是偶发性失败，问题当天查不出结果，等第二天复现时日志已经被清理了，只能从头再来。我建议对核心作业开启spark.eventLog.enabled=true，把事件日志做历史归档，事后回灌到 History Server 里复盘。

第二，提交脚本一定要版本化。不要小看这一步，我遇到过太多次"昨天还能跑、今天挂了"，最后发现是某个同事改了提交脚本里的内存参数。把提交命令、参数文件都放进 Git，每次变更留痕。

第三，建立自己的失败模板库。每排查完一个问题，就把报错信息、根因、复现条件、解决方案记下来。累积到几十条之后，你会发现自己排查新问题的速度提了一大截——因为大部分"新问题"只是"旧模板"的变体。

Spark 作业排查本身没有银弹，它考验的是你能否在多层日志、多个组件边界里保持清晰的判断顺序。先把报错分层，再从 UI 找线索，最后用最小化复现验证假设——这套流程坚持下去，凌晨三点盯着红色报错发呆的场景，会越来越少见。