1. 为什么Spark性能优化总在“异常”上翻车做数据开发这些年Spark任务跑得慢、跑挂、跑出各种稀奇古怪的报错几乎是每个从业者都绕不开的日常。很多人一上来就想着调参、加资源、改并行度结果折腾半天发现根本问题不在参数上而是某个异常反复出现却没有被真正理解。我见过太多团队任务一报错就去搜“Spark OOM怎么解决”搜到一堆复制粘贴的答案改完发现该挂还是挂。这篇内容聚焦的是Spark性能优化中常见异常的解决思路属于系列的第二篇。第一篇聊的是基础调优手段这一篇专门啃硬骨头——那些让你半夜被报警叫醒的异常。适合谁看如果你已经写过一段时间Spark任务能看懂DAG图知道Stage和Task的区别但遇到异常还是靠猜和试那这篇就是给你准备的。如果你刚入门也可以先了解这些异常长什么样以后遇到了不至于慌。核心关键词就几个数据倾斜、内存溢出、序列化异常、Shuffle失败、Executor丢失。这几个词背后对应的是Spark运行过程中最典型的几类故障。我不会只告诉你“调大内存”这种废话而是把每个异常的触发原理、排查路径、解决方案讲透让你下次遇到同类问题时能自己定位、自己解决。先说一个基本认知Spark的异常大多数不是孤立发生的。一个OOM可能引发Executor丢失Executor丢失又导致Shuffle文件拉取失败最后表现为任务反复重试直到超时。所以排查异常的第一步永远是看完整日志链路而不是盯着最后一个报错看。这个习惯我用了好几年帮我省了无数时间。2. 数据倾斜最磨人的性能杀手2.1 数据倾斜到底是怎么发生的数据倾斜的本质是某个或某几个Key对应的数据量远超其他Key导致处理这些Key的Task成为整个Stage的瓶颈。举个例子你有一张订单表按城市分组统计结果发现某个城市的订单量是其他城市的几百倍那这个城市对应的Reduce Task就要处理几百倍的数据量其他Task早就跑完了它还在慢慢磨。这种情况在Spark里表现为Stage进度卡在99%不动或者某个Task的耗时远超其他Task。你去看Spark UI的Stage页面会发现大部分Task几秒钟就完成了但有几个Task跑了十几分钟甚至更久。这就是典型的数据倾斜信号。数据倾斜的根源通常来自三个方面。第一是业务数据本身分布不均比如某些热门商品、头部用户天然占据大量数据。第二是Join操作时Key的分布不均大表和小表Join时如果小表的Key集中在少数几个值上就会导致倾斜。第三是GroupBy或Distinct操作如果分组字段的基数很低某些分组的数据量就会特别大。2.2 定位数据倾斜的实操方法定位倾斜不能靠猜得有数据支撑。我常用的方法有这么几个方法一看Spark UI的Task耗时分布。进入Stage详情页看Duration那一列如果最大值和最小值差距在10倍以上基本可以确认有倾斜。再看Shuffle Read Size倾斜Task的读取量通常也是最大的。方法二采样统计Key的分布。对怀疑有倾斜的字段做采样统计每个Key的出现次数。比如df.sample(0.1).groupBy(key_column).count().orderBy(count, ascendingFalse).show(20)这样能快速看出哪些Key的数据量异常大。方法三在日志里加计数器。在代码里对每个分区的数据量做统计输出到日志里。这个方法稍微麻烦一点但对复杂任务特别有效。提示定位倾斜时不要只看一个Stage要沿着DAG链路往上看。有时候倾斜发生在上游Stage但表现却在下游。2.3 解决数据倾斜的五种实战方案方案一加盐打散。这是最通用的方案。给倾斜的Key加上随机前缀让它分散到多个Task上处理处理完再去掉前缀做二次聚合。具体操作是先找出倾斜的Key然后对这些Key加随机后缀比如0到9这样原来一个Key就变成了十个Key分散到十个Task上。处理完后再做一次聚合把结果合并。这个方案的关键在于随机后缀的数量。太少起不到打散效果太多会增加Shuffle开销。我的经验是如果某个Key的数据量是平均值的N倍那随机后缀的数量至少要是N的平方根。比如某个Key的数据量是平均值的100倍那至少加10个后缀。方案二Map Join。如果倾斜发生在Join操作中且其中一张表足够小默认阈值是10MB可以通过spark.sql.autoBroadcastJoinThreshold调整那就把小表广播到每个Executor在Map阶段完成Join完全避免Shuffle。这是最彻底的方案但前提是小表真的能放进内存。方案三拆分倾斜Key单独处理。把倾斜的Key单独拿出来处理剩下的Key正常处理最后合并结果。比如你知道某个城市的订单量特别大那就把那个城市的数据单独拎出来跑一个任务其他城市跑另一个任务。这个方案适合倾斜Key数量少且明确的情况。方案四调整并行度。增加Shuffle Read的并行度让更多Task来分担数据。可以通过spark.sql.shuffle.partitions来调整。但要注意并行度不是越大越好每个Task的启动和调度都有开销。一般来说并行度设置为Executor核数的2到3倍比较合适。方案五使用两阶段聚合。先做局部聚合再做全局聚合。比如先按“Key随机前缀”做一次聚合然后再按Key做一次聚合。这样第一次聚合就把数据量降下来了第二次聚合的压力就小很多。方案适用场景优点缺点加盐打散通用倾斜适用范围广需要二次聚合Map Join大小表Join彻底消除Shuffle小表必须能放进内存拆分Key倾斜Key少且明确针对性强需要手动维护Key列表调整并行度轻度倾斜操作简单效果有限两阶段聚合GroupBy倾斜效果稳定代码改动较大2.4 一个真实的数据倾斜排查案例之前有个任务每天凌晨跑平时20分钟就能跑完突然有一天跑了3个小时还没结束。看Spark UI发现某个Stage卡在99%有几个Task跑了两个多小时。去看Shuffle Read Size发现最大的Task读了将近20GB数据而其他Task只读了几百MB。采样统计后发现某个字段有一个值的数据量是其他值的上千倍。这个值是业务系统的一个默认值当天因为上游系统的问题大量数据都被写成了这个默认值。解决方案是加盐打散给这个默认值加上0到99的随机后缀把数据分散到100个Task上。改完之后任务恢复到了25分钟。这个案例的教训是数据倾斜不一定是代码问题也可能是数据问题。排查时要先看数据分布再决定用什么方案。3. 内存溢出OOM的三种面孔和对应解法3.1 Driver OOM和Executor OOM的区别很多人一看到OOM就懵了其实OOM分两种Driver端和Executor端。这两者的原因和解决方案完全不同。Driver OOM通常发生在收集大量数据到Driver端的时候。比如你用了collect()、toPandas()、broadcast()这些操作把大量数据拉到了Driver。Driver的内存默认是1GB左右稍微大一点的数据集就扛不住。解决方案很简单能不用collect就不用非要用就分批拉取或者调大spark.driver.memory。Executor OOM更常见原因也更多样。可能是单个Task处理的数据量太大可能是内存中缓存了太多数据也可能是Shuffle操作时内存不够用。Executor OOM的报错信息里通常会提到“GC overhead limit exceeded”或者“Java heap space”。3.2 Executor OOM的排查路径排查Executor OOM我一般按这个顺序来第一步看是哪个Stage出的问题。如果是Shuffle阶段的OOM通常是Shuffle数据量太大。如果是Map阶段的OOM可能是单个分区数据量太大。第二步看Task的输入数据量。在Spark UI里找到出错的Stage看Task的Input Size和Shuffle Read Size。如果某个Task的输入数据量远超其他Task那就是数据倾斜导致的OOM解决方案参考上一章。第三步看内存使用情况。在Spark UI的Executor页面看每个Executor的Memory Used和GC Time。如果GC Time占比很高超过10%说明内存压力大需要调整内存配置。第四步检查代码里有没有内存泄漏。比如在循环里不断创建大对象、缓存了不该缓存的数据、使用了低效的数据结构等。3.3 内存配置的实操建议Spark的内存模型分三块Execution Memory、Storage Memory、User Memory。Execution Memory用于Shuffle、Join、Sort等操作Storage Memory用于缓存数据User Memory用于用户自定义的数据结构和UDF。默认情况下Execution和Storage共享一个内存池可以互相借用。这个比例由spark.memory.fraction控制默认是0.6意思是60%的堆内存用于Execution和Storage剩下的40%用于User Memory和系统开销。调优时我通常这样配置spark.executor.memory8g spark.executor.memoryOverhead2g spark.memory.fraction0.6 spark.memory.storageFraction0.5 spark.executor.cores4memoryOverhead是堆外内存用于JVM自身开销、网络缓冲等。经验值是executor.memory的10%到15%但如果你用了Python UDF或者大量网络传输可以调到20%。注意executor.memory不是越大越好。单个Executor内存太大GC时间会变长。一般来说单个Executor的内存不要超过32GB超过的话GC停顿会很明显。3.4 减少内存使用的代码技巧除了调参数代码层面的优化同样重要。我常用的技巧有用reduceByKey替代groupByKey。groupByKey会把所有数据拉到一起再聚合reduceByKey会先在Map端做局部聚合数据量能减少很多。用mapPartitions替代map。mapPartitions可以按分区处理数据减少对象创建的开销。比如你要对每条数据做一个耗时的初始化操作用mapPartitions只需要每个分区初始化一次。及时释放不再使用的DataFrame。用unpersist()释放缓存的DataFrame用df.unpersist()或者spark.catalog.clearCache()清理缓存。避免在循环里创建DataFrame。每次创建DataFrame都会生成新的执行计划循环里创建会导致执行计划膨胀Driver内存飙升。使用Kryo序列化。默认的Java序列化效率低、体积大。设置spark.serializerorg.apache.spark.serializer.KryoSerializer并注册自定义类能显著减少内存占用和网络传输量。4. 序列化异常容易被忽视的隐形杀手4.1 序列化异常为什么难排查序列化异常的特点是报错信息往往很模糊比如“Task not serializable”或者“java.io.NotSerializableException”但不告诉你具体是哪个对象不可序列化。而且这个问题经常在本地测试时不出现一上集群就报错让人很头疼。根本原因是Spark的分布式计算模型Driver端的代码会被序列化后发送到Executor执行。如果代码里引用了不可序列化的对象比如数据库连接、文件句柄、非序列化的自定义类就会在序列化时失败。4.2 常见的序列化陷阱陷阱一在闭包里引用了Driver端的对象。比如你在map操作里引用了一个数据库连接对象这个连接对象是不可序列化的就会报错。解决方案是把连接对象的创建放到Executor端用mapPartitions在每个分区里创建连接。陷阱二自定义类没有实现Serializable接口。如果你定义了一个类在RDD操作里用到了它这个类必须实现java.io.Serializable接口。而且类里的所有字段也必须是可序列化的。陷阱三在闭包里引用了外部类的成员变量。在Scala里如果在闭包里引用了外部类的成员变量整个外部类都会被序列化。如果外部类不可序列化就会报错。解决方案是把需要的变量提取到局部变量里。陷阱四使用了不可序列化的第三方库对象。比如某些HTTP客户端、连接池对象等。解决方案是在Executor端创建这些对象而不是在Driver端创建后传过去。4.3 排查序列化异常的实操步骤第一步看完整的报错堆栈。报错信息里通常会有一行“Object not serializable”后面跟着类名。找到这个类名看它为什么不可序列化。第二步检查闭包里引用的所有对象。把map、filter、flatMap等操作里引用的所有外部对象列出来逐个检查是否可序列化。第三步用序列化测试工具验证。可以写一个简单的测试用ObjectOutputStream尝试序列化可疑对象看是否报错。第四步逐步缩小范围。如果找不到具体是哪个对象可以把闭包里的代码逐步注释掉看哪部分去掉后不再报错。4.4 序列化优化的最佳实践使用Kryo序列化。Kryo比Java序列化快很多体积也小很多。配置方式是spark.conf.set(spark.serializer, org.apache.spark.serializer.KryoSerializer) spark.conf.set(spark.kryo.registrationRequired, false)如果开启了registrationRequired需要注册所有自定义类。不开启的话Kryo会自动处理但性能稍差。避免在闭包里引用大对象。闭包会被序列化后发送到Executor如果闭包里引用了一个很大的对象序列化和网络传输的开销都会很大。解决方案是把大对象拆分成小对象或者用广播变量。使用广播变量共享只读数据。如果多个Task需要用到同一份数据用广播变量比在每个Task里单独创建要高效得多。广播变量会被缓存在每个Executor上不会重复传输。自定义类尽量简单。自定义类里不要放太多字段不要放不可序列化的字段。如果某个字段不需要序列化可以用transient注解标记。5. Shuffle失败与Executor丢失的连锁反应5.1 Shuffle失败的常见原因Shuffle是Spark里最昂贵的操作也是最容易出问题的地方。Shuffle失败的原因主要有这么几类磁盘空间不足。Shuffle过程中会把中间数据写到磁盘如果磁盘空间不够就会报“No space left on device”。这个错误很直接清理磁盘或者增加磁盘空间就行。Shuffle文件丢失。如果某个Executor挂了它写的Shuffle文件就找不到了。下游Task去拉取这些文件时会报“FetchFailedException”。这种情况通常需要重跑上游Stage。网络超时。Shuffle数据量太大网络传输超时。可以调大spark.network.timeout和spark.shuffle.io.connectionTimeout。Shuffle服务未启动。如果用的是外部Shuffle服务服务没启动或者挂了也会导致Shuffle失败。5.2 Executor丢失的排查思路Executor丢失通常表现为任务运行过程中Executor数量突然减少任务变慢或者失败。排查思路如下第一步看Executor日志。找到丢失的Executor的日志看最后几行是什么。常见的原因有OOM被YARN杀掉、GC时间过长被心跳超时踢掉、磁盘满了等。第二步看YARN或K8s的事件日志。如果是YARN看ResourceManager的日志搜索“Container killed”相关的信息。如果是K8s看Pod的事件。第三步检查内存配置。如果Executor是因为OOM被杀需要调大executor.memory或者memoryOverhead。如果是GC导致的需要优化代码减少内存占用。第四步检查数据倾斜。如果某个Executor处理的数据量特别大也容易被压垮。参考第2章的倾斜解决方案。5.3 提高Shuffle稳定性的配置spark.shuffle.file.buffer64k spark.reducer.maxSizeInFlight96m spark.shuffle.io.maxRetries10 spark.shuffle.io.retryWait10s spark.shuffle.service.enabledtrue spark.network.timeout800s spark.executor.heartbeatInterval60s这几个参数的作用分别是增大Shuffle写缓冲区、增大Reduce端拉取数据的内存、增加重试次数、延长重试间隔、启用外部Shuffle服务、延长网络超时、延长心跳间隔。提示spark.executor.heartbeatInterval要小于spark.network.timeout否则Executor还没发心跳就被判定为超时了。一般心跳间隔是超时时间的十分之一左右。5.4 一个Shuffle失败的排查实录有个任务跑了几个小时突然报了一堆FetchFailedException。看日志发现是某个Executor挂了它写的Shuffle文件找不到了。进一步排查发现这个Executor是因为磁盘满了被YARN杀掉的。为什么磁盘会满因为这个任务的Shuffle数据量特别大而且没有启用外部Shuffle服务Shuffle数据都写在了Executor的本地磁盘上。解决方案有两个一是启用外部Shuffle服务把Shuffle数据写到专门的磁盘上二是增加Executor的磁盘空间。最后我们启用了外部Shuffle服务并且把spark.shuffle.file.buffer从默认的32k调到了64k任务稳定了很多。这个案例的教训是Shuffle相关的配置不能全用默认值要根据数据量做调整。6. 异常排查的通用方法论与工具6.1 建立系统化的排查思维排查Spark异常最忌讳的就是东一榔头西一棒子。我总结了一个通用的排查框架分四步第一步确定异常的类型和范围。是OOM还是序列化异常是单个Task失败还是整个Stage失败是偶发还是必现这些信息决定了排查方向。第二步收集完整的日志和指标。包括Driver日志、Executor日志、Spark UI的Stage和Task详情、YARN或K8s的事件日志。信息越全定位越快。第三步从现象反推原因。比如看到FetchFailedException就要想到可能是Executor挂了或者网络问题。看到GC overhead limit exceeded就要想到内存不够或者数据倾斜。第四步验证假设并修复。找到可能的原因后通过调整配置或修改代码来验证。如果问题解决了说明假设正确如果没解决继续排查。6.2 常用排查工具和技巧Spark UI是首选工具。通过Spark UI可以看到Stage的进度、Task的耗时分布、Shuffle的读写量、Executor的内存和GC情况。我一般会重点关注这几个页面Stages页面看整体进度Stage详情页看Task分布Executors页面看内存和GC。日志聚合工具。如果集群规模大用YARN的日志聚合或者ELK之类的工具来收集和搜索日志。关键是能按Executor ID或者Task ID过滤日志。Spark的Event Log。开启spark.eventLog.enabledtrueSpark会把所有事件写到日志文件里。用Spark History Server可以回放整个任务的执行过程对排查偶发问题特别有用。自定义Metrics。在代码里加一些自定义的计数器比如统计每个分区的数据量、每个Key的出现次数等。这些信息在排查倾斜和OOM时特别有用。6.3 常见异常速查表异常信息可能原因排查方向解决方案OutOfMemoryError: Java heap space内存不足或数据倾斜看Task输入数据量、GC时间调大内存、解决倾斜、优化代码GC overhead limit exceeded内存压力大、GC频繁看GC时间占比调大内存、减少缓存、优化数据结构Task not serializable闭包引用了不可序列化对象检查闭包里的所有引用用mapPartitions、提取局部变量、实现SerializableFetchFailedExceptionExecutor挂了或网络问题看Executor日志、网络配置调大超时、启用外部Shuffle服务、增加重试Container killed by YARN内存超限或磁盘满看YARN日志、磁盘使用率调大内存、清理磁盘、解决倾斜No space left on device磁盘空间不足看磁盘使用率清理磁盘、增加磁盘、启用外部ShuffleShuffle file not foundExecutor挂了导致Shuffle文件丢失看Executor状态重跑上游Stage、启用外部Shuffle服务6.4 预防胜于治疗日常开发习惯与其等出了问题再排查不如在日常开发中就养成好习惯。我自己的习惯有这么几条开发阶段就用小数据集测试。不要等上了集群才发现问题。用小数据集跑一遍能提前发现序列化异常、代码逻辑错误等问题。代码里加足够的日志。关键步骤加日志比如每个Stage的开始和结束、每个分区的数据量、关键变量的值等。出问题时这些日志就是救命稻草。配置参数不要全用默认值。根据数据量和集群规模调整关键参数比如并行度、内存、Shuffle相关配置。但也不要盲目调参每次只调一个参数观察效果。定期检查集群状态。看磁盘使用率、内存使用率、Executor的GC情况等。提前发现潜在问题避免任务跑到一半挂掉。保留历史任务的配置和日志。出问题时可以对比正常任务和异常任务的配置差异快速定位问题。7. 几个容易被忽略的细节7.1 动态资源分配的双刃剑spark.dynamicAllocation.enabledtrue可以让Spark根据负载动态增减Executor看起来很美但用不好会出问题。比如任务刚开始时Executor很少Shuffle数据写得到处都是后面Executor增加了又要去拉取之前写的Shuffle数据网络开销很大。我的建议是如果任务的数据量比较稳定关掉动态资源分配用固定数量的Executor。如果数据量波动大可以开启但要设置合理的minExecutors和maxExecutors并且启用外部Shuffle服务。7.2 推测执行的适用场景spark.speculationtrue可以开启推测执行对慢Task启动一个副本谁先完成用谁的结果。这个功能对数据倾斜导致的慢Task有一定效果但也会增加资源消耗。我一般只在任务对延迟敏感、且资源充足的情况下开启推测执行。如果资源紧张开启推测执行反而会拖慢整体进度因为副本Task会抢占资源。7.3 数据本地性的影响Spark会尽量把Task调度到数据所在的节点上这就是数据本地性。本地性级别从高到低是PROCESS_LOCAL、NODE_LOCAL、RACK_LOCAL、ANY。如果本地性级别太低Task需要跨节点拉取数据网络开销会很大。可以通过spark.locality.wait来控制等待时间。默认是3秒如果网络带宽充足可以调小如果网络是瓶颈可以调大。但等待时间太长也会导致Task调度延迟需要权衡。7.4 小文件问题的处理如果输入数据有很多小文件每个文件都会生成一个分区导致Task数量过多调度开销大。解决方案是用spark.sql.files.maxPartitionBytes控制每个分区的最大字节数或者用coalesce()和repartition()合并分区。输出时也要注意小文件问题。如果每个分区都写一个文件最终会产生大量小文件。可以用repartition()减少分区数或者在写入时用coalesce()合并。我在实际使用中发现Spark的异常排查能力是随着经验积累逐步提升的。刚开始遇到OOM就只知道加内存遇到倾斜就只知道加并行度效果往往不好。后来慢慢学会了看日志、看指标、分析数据分布才能对症下药。踩过几次坑之后我养成了一个习惯每次任务出问题不管最后解没解决都把排查过程和解决方案记下来。时间长了就形成了一套自己的排查手册再遇到类似问题时能快速定位。最后再分享一个小技巧如果你不确定某个配置改了之后会不会有效果可以先在一个小规模的数据集上测试对比修改前后的执行时间和资源消耗。这样比直接上生产环境试错要安全得多。