企业官网建设流程全解析

周五晚上十一点半，我盯着Spark UI上那一片红得刺眼的Stage列表，第22个stage已经重试了三次，每次attempt都停在同一个任务上。同事在旁边问：“是不是executor内存不够？要不把--executor-memory加到16g？”我说先等等，这种判断顺序搞反了。

这篇博文想聊的就是“Spark程序运行常见错误解决方法以及优化”。我见过很多生产环境，绝大多数Spark问题都逃不出这几个大类：内存与Container被杀、闭包序列化失败、Shuffle文件与executor丢失、数据倾斜导致的隐性失败，最后才是纯逻辑Bug。盲目调参数是最常见的一种“努力”，却往往偏离真实故障点。

适合谁看？刚把Spark跑起来、经常被一堆红色日志吓到的新手；已经能跑任务但偶尔深夜被集群报警叫醒的开发者；以及想系统整理一遍Spark排障思路的数据工程师。下文没有高深理论，全是可以在集群上直接对照检查的操作经验和参数建议。

1. 先别急着改参数：把“Stage失败”翻译成具体故障类型

1.1 三条线索快速判断故障层

Spark的报错是分层的。顶层永远是那一句“Job aborted due to stage failure”，这行字除了告诉你“某个stage失败了”之外，什么诊断价值都没有。真正有用的信息藏在三层往下：

• 用户代码异常：栈顶出现你的类名，比如java.lang.NullPointerException或java.lang.ClassCastException，这类问题基本都是业务逻辑或类型转换错误，属于代码层故障。
• JVM进程级异常：OutOfMemoryError、GC overhead limit exceeded、Container killed by YARN for exceeding memory limits，这类问题指向executor/JVM或容器资源限制。
• 通信与调度异常：ExecutorLostFailure、Could not find executor、FetchFailedException、Connection refused，这类问题指向集群调度、网络、shuffle中间文件等基础设施。

为什么要先分清这个？因为处理方式完全不同。代码层的问题改代码，资源层的问题调内存和并行度，基础设施层的问题要看集群配置。很多新手拿到报错第一反应是“是不是我代码写错了”，第二反应是“是不是参数不够大”，这其实跳过了最重要的一步：确认故障在哪个层。

1.2 定位日志要看的四个位置

在终端里翻那个大红堆栈，是最慢的方式。按顺序查这四处，效率会高很多：

1. Spark Web UI的Stages页面：看是哪个stage失败、失败任务数、任务被重试了几次。几十个任务里只有一两个失败，和几百个任务全部失败，含义完全不同。
2. executor的stderr日志：任务级异常、OOM信息、GC日志基本都在这里。注意是stderr而不是stdout，两者混淆是新手常见动作。日志位置一般在YARN的container日志目录里，Executors页面也有Logs链接可以直接点进去。
3. YARN资源管理器页面：确认对应application下的container是否被删除、被杀，“Diagnostics”栏会直接给出被杀原因，比如内存超限或者物理内存超标。
4. driver日志：如果driver端直接异常，任务可能压根没提交成功，或者collect()拉回的数据太大导致driver OOM。

我见过太多人跳过1和2，直接搜报错串，然后在几十个相似但实际不对的博客答案里浪费时间。其实Spark UI自己已经帮你在executor日志里标好了“cause”字段，大部分问题在第一屏就能判断方向。

把故障分好类，下一步就得处理最常见的那个大类：内存。

2. Container killed 不是代码写错了：Spark内存模型和进程被杀的真实链路

2.1 Spark对“内存爆了”的三种解释

“executor内存不够”这句话其实很模糊。Spark进程里的内存被切成好几块，出问题时的现象很像，但处理方法完全不同：

所以“内存爆了”常见有三种形态。第一种是Execution内存不足，多数表现为磁盘spill、任务变慢但没死；如果spill还兜不住，就可能GC爆炸或OOM。第二种是User内存炸了，典型场景是某个task里构造了一个几GB的HashMap，任务一启动就OOM，这是代码层面的锅。第三种是堆外内存超了，很多人忽略overhead，结果YARN直接杀了container，这就是著名的“Container killed by YARN for exceeding memory limits”问题。

一个关键认知：YARN决定杀进程的依据，是容器内进程的整体物理内存RSS，不是JVM堆的大小。就算你给executor配了8g堆，netty的direct buffer、线程栈、JNI开销都会把RSS推到超过container阈值。所以堆设置正常不代表不会被杀。

2.2 一份减少内存溢出的可操作检查表

我在某公司遇到过一个“跑批任务一到凌晨就挂”的问题。现象是任务跑了40分钟，shuffle完成后开始大量executor被杀。UI里看到Storage页面缓存了接近3GB的DataFrame，executor的平均GC时间超过30%。

当时的排查顺序是这样的：

1. 先不看参数，看代码：发现任务把一张不小的事实表cache了一整份，但后面其实只复用了一次。这个cache就是storage内存的元凶。把cache改成只在确实多次复用的节点上做，并且用MEMORY_AND_DISK_SER，storage内存占用直接从3GB降到几百MB，进程GC压力明显缓解。
2. 调整executor内存配比：spark.memory.fraction从0.6提到0.75，同时storageFraction从0.5降到0.3，因为该任务根本没什么缓存。
3. 最后检查overhead：executor是8g，默认overhead按10%算约819MB，已经够用，不需要加。

另一个容易被忽视的规则：executor-cores越大，同时间在一个executor内跑的task越多，每个task分到的内存就越少。所以不要盲目“CPU核数越多越好”，很多OOM其实不是内存不够大，而是并发task太多，把Executor内存切碎了。一个保守的经验值是单executor给2~4个核，内存和核数配比维持每个核吃到2g~3g的堆。

检查清单：

• [ ] 确认是不是缓存把storage区域塞满了（UI的Storage页）
• [ ] 确认是不是shuffle缓冲把execution区域撑爆（GC日志中出现大量spill、GC overhead）
• [ ] 确认是不是代码里建了大结构（一个task巨慢且只影响少数executor）
• [ ] 确认RSS是否超出container上限（YARN diagnostics会直接说）
• [ ] 逐个排除后再动参数，不要上来就加executor-memory

内存问题聊完了，另一种同样高频的报错跟内存无关，它出在代码向executor传递对象的环节。

3. NotSerializableException 的真相：闭包把你不想带的对象抓进去了

3.1 一个最容易被忽略的Spark细节：Task的序列化传递

Spark分布式计算的基本单元是task。Driver上定义的函数、过滤器、映射逻辑，最终会被打包发送给executor执行。这个“打包”动作需要序列化——函数本身没问题，但函数内部引用了外部对象时，Spark会把整个依赖链全部序列化后塞进task。

于是经典报错出现了：org.apache.spark.SparkException: Task not serializable，Cause: java.io.NotSerializableException。

很多人困惑“我的类明明没进Spark逻辑，为什么报错”。举一个我常见到的场景：

// 报错版本 class UserService { val client = new SomeHeavyClient() // 包含了数据库连接，不可序列化 def process(df: DataFrame): DataFrame = { df.map(row => { client.query(row.getString(0)) // 匿名函数捕获了 this.client row }) } }

这里的row => { client.query(...) }是定义在UserService实例的方法内部，匿名函数自动捕获了外层的this，也就是整个UserService对象，而它内部的client不可序列化，于是一提交task就炸。

很多人的第一反应是给UserService加extends Serializable，但实际上数据库连接这种资源型对象序列化过去也毫无意义——executor上反序列化出来的连接并不等于一个可用连接，这是个伪修复。

3.2 从报错到修复：一个真实代码场景

正确做法有几种，按推荐顺序：

• 方法一：别让闭包直接访问外部实例字段。把需要的参数作为独立变量传入，尽量只捕获基本类型或String。
• 方法二：把不可序列化的资源改成executor端惰性初始化（懒加载单例）。在算子内部定义if (client == null) client = createClient()，让连接在executor本地创建，不在driver创建。
• 方法三：把逻辑塞进一个object伴生对象里，避免实例状态被隐式捕获。
• 方法四：开启Kryo序列化，spark.serializer=org.apache.spark.serializer.KryoSerializer，并在参数里注册需要的类。Kryo对常见类序列化更快更小，能解决很多“明明实现了Serializable却仍报错”的边角问题。

修完之后的版本：

class UserService { def process(df: DataFrame): DataFrame = { df.mapPartitions(iter => { val client = new SomeHeavyClient() // 在executor端创建 iter.map(row => { client.query(row.getString(0)); row }) }) } }

注意我把map改成了mapPartitions，这样连接在整个分区内只创建一次，不会每条记录都new一个——顺手把性能问题也解决了。

这里有一个我每次排障都会强调的点：遇到NotSerializableException，先看“闭包捕获了什么”，而不是先加Serializable。绝大多数情况下要么是捕获了不该捕获的对象，要么是UDF里放了一个resource型字段。

序列化问题大多出在代码层，接下来看一个涉及集群协调层的经典错误。

4. Shuffle 拉取失败：稳定运行的另一半靠的是中间文件不丢

4.1 为什么Map端的文件一出问题，Reduce端就会连环失败

Shuffle是Spark里最“重”的一步：map端把中间结果写到本地磁盘，reduce端需要时再从对方的BlockManager拉取。这个过程中，中间文件一旦丢失，或者存放文件的executor没了，reduce端就会抛出FetchFailedException、Connection refused、Could not find executor之类的一连串错误。

为什么叫“连环”？因为一个reduce任务要拉几百个map输出文件，只要其中一个源文件不可达，整个reduce任务就失败；如果该源executor还没法恢复，stage可能会被重试。但重试时，如果还是同一个链路，依然失败。这就是为什么很多任务会“重试三次再整体失败”。

最常见的三种触发条件：

1. shuffle文件所在的executor因OOM被杀，文件随之消失。这是内存问题蔓延成shuffle问题的典型路径。
2. 动态资源分配把空闲executor回收得太快，reduce任务还没拉完它的map输出，executor已经被释放，文件没了。
3. 本地磁盘不足，map端写shuffle文件失败，或者写了一半进程崩溃。

4.2 外部Shuffle服务与关键网络参数

针对上述原因，最有效的一招是开启external shuffle service（spark.shuffle.service.enabled=true）。外部Shuffle服务独立于executor运行，它直接管理executor写出的shuffle文件。executor和container被回收后，文件仍在，新的reduce端依然能拉。它不能解决所有问题，但能解决最常发生的“executor没了文件也没了”这种情况。

如果开了这个服务还是偶尔出现FetchFailed，再检查这几个参数：

• spark.network.timeout：默认120秒，如果网络抖动频繁、节点负载高导致响应超时，适当调大到300~600秒。
• spark.shuffle.io.retryWait：默认5秒，重试期间的退避间隔。FetchFailed后会重新选择目标executor，给足重试时间很关键。
• spark.shuffle.io.maxRetries：默认3，网络条件差的环境可以到5~6。
• spark.sql.adaptive.shuffle.targetPostShuffleInputSize：配合AES启用，控制reduce端每个分区预期处理的数据量，能很好规避部分executor shuffle read过大的问题。

还有两个容易被忽略的点。一是本地目录：给Spark配置多个spark.local.dir并分散在不同磁盘，能避免单个盘写满导致shuffle写盘失败。二是不要把动态分配和external shuffle service割裂开考虑：动态分配下executor频繁波动，shuffle tracking一定要开启（Spark 3.x在动态分配启用时默认跟随生效，但需要确认）。

Shuffle中间文件的问题是稳定性的一半，另一半则是数据本身的分布问题，它就是数据倾斜。

5. 数据倾斜不会立刻报错，却会让任务悄悄“卡死”

5.1 怎么在Spark UI里一眼认出数据倾斜

数据倾斜不算严格意义上的“运行错误”，但它比错误更难对付——任务不报错，只是某个stage里一两个task慢得出奇，其余task早就结束了。如果数据倾斜极端，个别task会OOM或者反复拉取失败，最终表现为“整个任务hang住然后失败”。

在Spark UI的Stages页面，看任务的持续时间和shuffle read字节数就能识别：大部分任务几秒钟完成，但某几个任务涨成几十倍的时间条；或者shuffle read量差异特别大，最大一个task读了几个GB，中位数只有几十MB。如果executor的GC时间也比平均值高出一大截，那基本就是倾斜没跑了。

还有一种隐蔽情况：同一个key被分到同一个reduce task，该task积累了大量数据，在groupBy/join时直接把execution内存打满，表现出来反而不是“慢”而是OOM。

倾斜有规律可循，通常出现在聚合、join、分区字段分布不均三种场景。下面给三种被验证有效的处理手段。

5.2 三种常用的倾斜处理手段

手段一：两阶段聚合（适用于groupBy/聚合倾斜）

先给key加随机前缀，把大key拆成多个子key，完成第一轮局部聚合；第二轮去掉前缀，再聚合一次。第二轮的key数量小了，压力自然分散。

// 第一阶段：加随机前缀，做局部聚合 val srcDF = ... val salted = srcDF.withColumn( "salt", when(col("key").isin(skewKeys.toSeq: _*), floor(rand() * 10).cast("int")) .otherwise(lit(0)) ) val r1 = salted .groupBy("key", "salt") .agg(sum("amount").as("partial_sum")) // 第二阶段：去掉前缀，做最终聚合 val r2 = r1 .groupBy("key") .agg(sum("partial_sum").as("total_sum"))

手段二：广播小表（适用于小表join大表倾斜）

如果被广播的表够小（默认spark.sql.autoBroadcastJoinThreshold=10MB），Spark会在map端完成关联，不产生reduce端的shuffle，从根本上避免倾斜。如果表在20MB~50MB，可以手动hint：

SELECT /*+ BROADCAST(small) */ ... FROM fact f JOIN dim d ON f.key = d.key

手段三：倾斜key单独处理

这是大表join大表倾斜时的正规方案：先把倾斜key识别出来，对倾斜key的数据单独加随机前缀后和另一张表的膨胀副本关联，再与非倾斜key的关联结果union。实现稍有复杂度，但效果立竿见影。原理说清楚，比背代码有用。

另外，如果你用的是Spark 3.x，AES里的optimizeSkewedJoin特性会自动切开倾斜分区的shuffle读取，优先开启它，很多时候不需要自己写两阶段聚合。

倾斜处理完，任务算稳定了，最后才是大多数人最关心的提速。

6. 从“能跑”到“跑得快”：一份可复制的参数与代码优化清单

6.1 稳定性优先的参数基调

参数建议分两层：先稳定，再提速。稳定性优先，因为没有稳定，一切性能都是零。

6.2 性能优化先看两个指标，再谈调整

我无数次被人问“怎么调优”，我的标准回答是：先看两个指标。

第一，shuffle总字节数。Shuffle Read和Shuffle Write是同一个值，它代表数据在集群里搬动了多少。如果这个值是几个GB甚至几十GB，先别急着调并发，先想想能不能从计算逻辑上减少shuffle：少用join、提前filter、用broadcast代替大表shuffle、避免groupByKey而用reduceByKey。

第二，任务拖尾情况。如果大部分task几十秒完成，少数task几分钟还没结束，那调任何参数都救不了，问题大概率是倾斜或资源竞争，要回到第5章的方案。如果时间分布平滑，才考虑并行度是否吃满。

一个经常被低估的代码优化是：避免无脑cache。cache本身不会加速计算，它只是把重复计算的代价换成存储代价。只有在一个数据集被多次复用、且复用之间还隔着较大计算量时才值得缓存。而且缓存建议用MEMORY_AND_DISK_SER，不要用MEMORY_ONLY，后者一旦放不下就直接驱逐，可能反而引发反复计算。

6.3 一个生产级spark-submit示例

spark-submit \ --master yarn \ --deploy-mode cluster \ --num-executors 50 \ --executor-cores 4 \ --executor-memory 8g \ --driver-memory 4g \ --conf spark.sql.shuffle.partitions=400 \ --conf spark.sql.adaptive.enabled=true \ --conf spark.shuffle.service.enabled=true \ --conf spark.dynamicAllocation.enabled=false \ --conf spark.serializer=org.apache.spark.serializer.KryoSerializer \ --conf spark.executor.extraJavaOptions="-XX:+UseG1GC -XX:MaxGCPauseMillis=200" \ --conf spark.network.timeout=300 \ --class com.example.MainApp your-app.jar

逐个说明：executor-cores=4、executor-memory=8g，简单算单核能分到2g堆，属于比较稳的比例。spark.sql.shuffle.partitions=400，和50个executor×4核=200个并发槽位基本匹配，避免一个槽位里task排队严重。关闭动态分配是为了夜间批处理任务的稳定性。G1GC比默认CMS更适合大堆和低停顿需求。Kryo和AES默认开启，这两项的收益几乎是白拿的。

优化之后不要只看“任务变快了多少”，要记录优化前的shuffle量、平均task时间、executor GC时间，改一次测一次，避免陷入参数玄学。

最后说一点我自己的习惯：每次遇到Spark任务问题，我先问三个问题——是哪个stage失败、那个stage在哪些task上失败、失败前executor日志里的最后几行是什么。90%的情况下，这三个问题能让我在十分钟内找到方向。剩下的十分钟，才是调参数和改代码。