Spark任务排查与性能优化：从OOM、Shuffle失败到数据倾斜

周五晚上十一点半我盯着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 定位日志要看的四个位置在终端里翻那个大红堆栈是最慢的方式。按顺序查这四处效率会高很多Spark Web UI的Stages页面看是哪个stage失败、失败任务数、任务被重试了几次。几十个任务里只有一两个失败和几百个任务全部失败含义完全不同。executor的stderr日志任务级异常、OOM信息、GC日志基本都在这里。注意是stderr而不是stdout两者混淆是新手常见动作。日志位置一般在YARN的container日志目录里Executors页面也有Logs链接可以直接点进去。YARN资源管理器页面确认对应application下的container是否被删除、被杀“Diagnostics”栏会直接给出被杀原因比如内存超限或者物理内存超标。driver日志如果driver端直接异常任务可能压根没提交成功或者collect()拉回的数据太大导致driver OOM。我见过太多人跳过1和2直接搜报错串然后在几十个相似但实际不对的博客答案里浪费时间。其实Spark UI自己已经帮你在executor日志里标好了“cause”字段大部分问题在第一屏就能判断方向。报错现象去哪个界面找线索最常见的根因少量task失败重试后成功Stages页面看任务失败原因数据倾斜、瞬时资源不足、网络抖动大量task同时失败YARN看container被杀时间内存超限、节点故障、shuffle文件丢失任务卡在某stage很久看task耗时分布数据倾斜、资源不足driver进程直接退出driver日志driver内存不足、collect返回数据过大把故障分好类下一步就得处理最常见的那个大类内存。2. Container killed 不是代码写错了Spark内存模型和进程被杀的真实链路2.1 Spark对“内存爆了”的三种解释“executor内存不够”这句话其实很模糊。Spark进程里的内存被切成好几块出问题时的现象很像但处理方法完全不同内存区域默认配置通俗解释Reserved Memory300MBSpark自己保留的可以不关心Execution Memory(总堆-300MB) × spark.memory.fraction(0.6) × (1 - storageFraction)运算过程中的临时数据、shuffle缓冲、聚合哈希表放这里Storage Memory(总堆-300MB) × 0.6 × storageFraction(0.5)cache/persist缓存、广播变量放这里User Memory(总堆-300MB) × 0.4用户创建的临时大对象、UDF内部结构等Off-heap/Overheadspark.executor.memoryOverhead堆外DirectBuffer、Netty、线程栈YARN看的是这个加堆的物理总量所以“内存爆了”常见有三种形态。第一种是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的DataFrameexecutor的平均GC时间超过30%。当时的排查顺序是这样的先不看参数看代码发现任务把一张不小的事实表cache了一整份但后面其实只复用了一次。这个cache就是storage内存的元凶。把cache改成只在确实多次复用的节点上做并且用MEMORY_AND_DISK_SERstorage内存占用直接从3GB降到几百MB进程GC压力明显缓解。调整executor内存配比spark.memory.fraction从0.6提到0.75同时storageFraction从0.5降到0.3因为该任务根本没什么缓存。最后检查overheadexecutor是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 serializableCause: 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.serializerorg.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可能会被重试。但重试时如果还是同一个链路依然失败。这就是为什么很多任务会“重试三次再整体失败”。最常见的三种触发条件shuffle文件所在的executor因OOM被杀文件随之消失。这是内存问题蔓延成shuffle问题的典型路径。动态资源分配把空闲executor回收得太快reduce任务还没拉完它的map输出executor已经被释放文件没了。本地磁盘不足map端写shuffle文件失败或者写了一半进程崩溃。4.2 外部Shuffle服务与关键网络参数针对上述原因最有效的一招是开启external shuffle servicespark.shuffle.service.enabledtrue。外部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.autoBroadcastJoinThreshold10MBSpark会在map端完成关联不产生reduce端的shuffle从根本上避免倾斜。如果表在20MB~50MB可以手动hintSELECT /* BROADCAST(small) */ ... FROM fact f JOIN dim d ON f.key d.key手段三倾斜key单独处理这是大表join大表倾斜时的正规方案先把倾斜key识别出来对倾斜key的数据单独加随机前缀后和另一张表的膨胀副本关联再与非倾斜key的关联结果union。实现稍有复杂度但效果立竿见影。原理说清楚比背代码有用。另外如果你用的是Spark 3.xAES里的optimizeSkewedJoin特性会自动切开倾斜分区的shuffle读取优先开启它很多时候不需要自己写两阶段聚合。场景方案注意点groupBy聚合倾斜两阶段聚合要预先识别倾斜key或用AES自动优化小表join大表广播join被广播的表不能无限大注意driver内存大表join大表倾斜key拆分随机前缀实现复杂先评估是否真有必要只想缓解增大shuffle分区数治标不治本只能摊薄单个task数据量倾斜处理完任务算稳定了最后才是大多数人最关心的提速。6. 从“能跑”到“跑得快”一份可复制的参数与代码优化清单6.1 稳定性优先的参数基调参数建议分两层先稳定再提速。稳定性优先因为没有稳定一切性能都是零。参数建议值作用spark.sql.shuffle.partitions200~500取决于数据量和executor数reduce端分区数太少容易OOM太多增加调度开销spark.network.timeout300~600网络抖动下降低fetch失败概率spark.executor.memoryOverhead不要低于默认堆外任务多时开到堆的15%~20%防止堆外内存超限被YARN杀containerspark.dynamicAllocation.enabled生产环境看情况夜间批处理建议关闭波动大时开频繁增加/删除executor会放大shuffle文件丢失风险spark.shuffle.service.enabledtrue保证executor退出后shuffle仍可读spark.sql.adaptive.enabledSpark 3.xtrue自动合并小分区、处理倾斜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.partitions400 \ --conf spark.sql.adaptive.enabledtrue \ --conf spark.shuffle.service.enabledtrue \ --conf spark.dynamicAllocation.enabledfalse \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer \ --conf spark.executor.extraJavaOptions-XX:UseG1GC -XX:MaxGCPauseMillis200 \ --conf spark.network.timeout300 \ --class com.example.MainApp your-app.jar逐个说明executor-cores4、executor-memory8g简单算单核能分到2g堆属于比较稳的比例。spark.sql.shuffle.partitions400和50个executor×4核200个并发槽位基本匹配避免一个槽位里task排队严重。关闭动态分配是为了夜间批处理任务的稳定性。G1GC比默认CMS更适合大堆和低停顿需求。Kryo和AES默认开启这两项的收益几乎是白拿的。优化之后不要只看“任务变快了多少”要记录优化前的shuffle量、平均task时间、executor GC时间改一次测一次避免陷入参数玄学。最后说一点我自己的习惯每次遇到Spark任务问题我先问三个问题——是哪个stage失败、那个stage在哪些task上失败、失败前executor日志里的最后几行是什么。90%的情况下这三个问题能让我在十分钟内找到方向。剩下的十分钟才是调参数和改代码。