1. 为什么Spark性能优化总在“异常”上翻车
干过几年大数据的人都有个共识：Spark作业跑不起来不可怕，可怕的是它跑得起来但跑得慢，或者跑着跑着突然给你甩一个异常堆栈，然后整个任务挂掉。更让人头疼的是，很多异常信息看起来像是资源不够，实际上根子出在数据倾斜或者序列化上。我见过太多团队一遇到OutOfMemoryError就无脑加内存，结果内存翻了一倍，任务该挂还是挂。
这篇内容接着上一期的思路，继续聊Spark性能优化中那些高频异常的处理方式。和上一期偏重“调参”不同，这一期我更想聚焦在异常背后的真实原因，以及怎么用最小的改动把问题摁住。适合已经能跑通Spark任务、但对异常排查还停留在“看日志猜原因”阶段的同学。如果你正在被java.lang.OutOfMemoryError、FetchFailedException、Task not serializable这几个老熟人反复折磨，那接下来的内容应该能帮你省下不少试错时间。
需要提前说明的是，Spark的版本差异对异常表现影响很大。我下面提到的现象和参数，主要基于Spark 2.4到3.x这个区间，更早的1.x版本有些行为不一样，遇到对不上的地方可以先确认下自己的版本。另外，所有案例都来自我经手过的模拟项目和公开技术社区的常见讨论，不涉及任何具体业务数据。
2. 内存类异常：别急着加内存，先搞清楚谁在吃内存
2.1 Driver端OOM和Executor端OOM是两码事
很多人一看到java.lang.OutOfMemoryError: Java heap space就条件反射地去调spark.executor.memory，但如果这个异常发生在Driver端，你调Executor内存调到天亮也没用。区分方法很简单：看异常堆栈里有没有driver相关的类，或者看Web UI里是哪个节点先挂的。
Driver端OOM的典型场景是collect()或者take()把大量数据拉回本地。我试过一个模拟项目，用collect()把一个几千万行的DataFrame拉到Driver做二次处理，结果Driver直接崩了。这种问题的正解不是加Driver内存，而是问自己一句：这些数据真的需要全部回到Driver吗？大多数情况下，用foreachPartition或者写回分布式存储再读，都比collect()靠谱。
Executor端OOM则更常见，原因也更多样。下面这张表是我自己排查时常用的对照思路：
| 异常表现 | 常见原因 | 优先排查方向 |
|---|---|---|
| Executor频繁GC后OOM | 单分区数据量过大 | 分区数、数据倾斜 |
| 堆内存溢出且堆栈指向序列化 | 对象太大或序列化方式低效 | 序列化器、数据结构 |
| 堆外内存溢出 | shuffle或缓存占用过多 | 内存管理模型、缓存策略 |
| 物理内存超限被YARN杀掉 | 堆外+堆内总和超限 | 内存开销系数 |
注意：YARN的
yarn.nodemanager.pmem-check-enabled如果开着，Executor的物理内存超限会被直接kill，日志里往往看不到OOM堆栈，只看到Container被kill。这种情况要去YARN的日志里找，别只盯着Spark的stderr。
2.2 内存管理模型决定了你该调哪个参数
Spark 1.6之后引入了统一内存管理模型，堆内内存被划分为三块：Execution内存、Storage内存和Other内存。Execution用于shuffle、join、sort等计算过程，Storage用于缓存RDD和广播变量。两者可以互相借用，但借用的前提是对方有空闲。
这个机制带来的一个坑是：缓存了大量数据后，Execution内存被挤占，导致shuffle时频繁spill甚至OOM。我遇到过好几次，任务里cache()了一个大表，前面跑得好好的，一到shuffle阶段就开始报Unable to acquire XXX bytes。解决办法要么是减少缓存量，要么是调大spark.memory.fraction，要么干脆用MEMORY_AND_DISK级别的存储策略，让缓存能溢写到磁盘。
具体参数上，spark.memory.fraction默认0.6，表示堆内60%给Execution和Storage共用。如果你的任务缓存需求大、shuffle需求也大，这个值可以适当提到0.7到0.8，但别超过0.8，否则Other内存不够，容易出各种奇怪的元数据问题。spark.memory.storageFraction默认0.5，表示共用区域里Storage占一半，这个值调大可以让缓存更稳，但会压缩Execution的空间。
2.3 堆外内存溢出：容易被忽略的“隐形杀手”
java.lang.OutOfMemoryError: Direct buffer memory这个异常出现频率不高，但一旦出现就很难缠。它通常和spark.memory.offHeap.enabled以及spark.memory.offHeap.size有关。堆外内存不受JVM堆大小限制，但受物理内存限制，而且YARN默认只检查物理内存，不检查堆外内存的分配细节。
我踩过的一个坑是：开了堆外内存做缓存，但没同步调整spark.executor.memoryOverhead。结果Executor的堆内内存看起来很正常，堆外却一直在涨，最后被YARN判定物理内存超限直接杀掉。后来我把memoryOverhead从默认的max(384MB, 0.1 * executorMemory)提到了0.15 * executorMemory，问题就消失了。
实操心得：如果你的任务用了堆外内存、Python UDF或者大量网络传输，
spark.executor.memoryOverhead建议直接给到0.15到0.2倍executor内存。别心疼那点内存，被kill一次重跑的成本远高于多占一点资源。
3. Shuffle相关异常：FetchFailed是结果，不是原因
3.1 FetchFailedException的三种典型触发路径
org.apache.spark.shuffle.FetchFailedException大概是Spark里最“声名在外”的异常之一。它的字面意思是Executor去拉取shuffle数据时失败了，但失败的原因可能完全不在拉取这个动作本身。我把它归纳为三条路径：
第一条是Map端输出丢失。Map任务所在的Executor因为内存超限被kill，或者磁盘写满导致shuffle文件写失败，下游Reduce去拉的时候自然拉不到。这种情况日志里通常能看到Failed to connect to /xxx:port或者Connection refused。
第二条是Shuffle文件过大导致拉取超时。单个shuffle块太大，网络传输时间超过spark.shuffle.io.connectionTimeout（默认120秒），就会报FetchFailed。这种情况在数据倾斜时特别常见，某个key对应的数据量是其他key的几十倍，Reduce端拉这个块拉到超时。
第三条是Executor本身不稳定。比如节点负载过高、GC停顿过长，导致心跳超时被Master判定为Lost，上面的shuffle数据也就跟着没了。
3.2 从参数和代码两个层面下手
针对第一条路径，核心是让Map端稳定。除了前面说的内存参数，还要关注spark.shuffle.file.buffer（默认32KB）和spark.shuffle.spill.compress。把file.buffer调到64KB或128KB，可以减少磁盘IO次数，降低Map端写失败的概率。spill.compress建议保持true，用snappy或lz4都行，压缩后的spill文件更小，写起来更快。
针对第二条路径，重点是缓解数据倾斜。我常用的手段有三个：一是加盐，给倾斜的key加上随机前缀，打散到多个分区；二是用repartition配合自定义分区器，把大key拆开；三是如果业务允许，直接过滤掉异常大的key单独处理。加盐这个操作听起来简单，但盐值的选择有讲究，一般取分区数的平方根到分区数之间比较合适，太小打散不够，太大增加shuffle开销。
针对第三条路径，可以调大spark.shuffle.io.maxRetries（默认3）和spark.shuffle.io.retryWait（默认5秒）。我一般会把重试次数提到5到10次，重试间隔提到10秒，给Executor一点恢复时间。同时把spark.network.timeout从默认120秒提到300秒甚至600秒，避免因为偶发的网络抖动就判定节点失联。
// 一个简单的加盐示例，给倾斜key加上0到9的随机前缀 val saltedDF = originalDF .withColumn("salted_key", concat(lit((Math.random() * 10).toInt), col("key"))) .repartition(200, col("salted_key"))注意：加盐之后，后续如果需要按原key聚合，记得把盐去掉再聚合一次，否则结果会翻倍。这个坑我见过不止一个团队踩过。
3.3 Shuffle文件清理与磁盘策略
Spark的shuffle文件默认在任务结束后由spark.shuffle.service.enabled配合外部shuffle服务来管理。如果没开外部shuffle服务，Executor退出时shuffle文件就没了，这也是为什么动态资源分配场景下必须开外部shuffle服务。
磁盘方面，spark.local.dir建议配置多个磁盘路径，用逗号分隔，让shuffle写操作分散到不同磁盘上。如果只有一块盘，那至少把spark.shuffle.spill.diskWriteBufferSize调大一点，减少写放大。另外，spark.cleaner.periodicGC.interval可以设成1min，让ContextCleaner更频繁地触发GC，及时释放不再使用的shuffle数据。
4. 序列化与闭包异常：那些“看起来不像性能问题”的性能问题
4.1 Task not serializable的根因与排查
org.apache.spark.SparkException: Task not serializable这个异常，表面上是序列化问题，实际上往往是因为在闭包里不小心引用了Driver端的对象。比如你在map里用了一个定义在Driver端的数据库连接，或者引用了一个不可序列化的工具类实例。
排查方法很直接：看异常堆栈里Serialization stack那一段，它会告诉你哪个对象不可序列化。常见的有java.sql.Connection、SimpleDateFormat、各种InputStream。解决办法要么是把这些对象在Executor端重新创建（用mapPartitions而不是map），要么把它们声明为transient并在使用前重新初始化。
我个人的习惯是，凡是需要在Executor端用的资源，一律用mapPartitions配合foreachPartition来创建和销毁，绝不在Driver端创建后往闭包里传。这样既避免了序列化问题，也避免了连接对象在多个Task之间共享带来的线程安全问题。
4.2 Kryo序列化：不是开了就万事大吉
spark.serializer设成KryoSerializer几乎是Spark调优的标配，但很多人开了之后发现性能提升不明显，甚至偶尔报KryoException。原因通常是没有注册自定义类。Kryo对未注册的类会退化成写类名的方式，序列化后的体积反而可能比Java序列化还大。
正确的做法是把任务里用到的自定义类都通过spark.kryo.registrator注册进去。如果类太多懒得一个个写，可以设spark.kryo.registrationRequired=false让Kryo自动处理，但这样会失去一部分性能优势。折中方案是只注册那些频繁序列化的核心类，比如自定义的POJO和枚举。
// 自定义Kryo注册器示例 class MyKryoRegistrator extends KryoRegistrator { override def registerClasses(kryo: Kryo): Unit = { kryo.register(classOf[MyCustomRecord]) kryo.register(classOf[MyEnum]) } } // 配置：spark.kryo.registrator=com.example.MyKryoRegistrator实操心得：Kryo的
spark.kryoserializer.buffer.max默认64MB，如果序列化大对象时报Buffer overflow，把它调到128MB或256MB。这个参数在广播大变量时特别容易触发。
4.3 闭包清理与广播变量
Spark的闭包清理机制（ClosureCleaner）会尝试把闭包里不需要的引用清理掉，但有时候它清理不干净，导致整个外部对象被序列化。一个典型场景是在map里引用了外部类的一个字段，而这个外部类本身很大。解决办法是用局部变量把需要的字段取出来，再在闭包里用局部变量。
广播变量是另一个容易出问题的地方。sc.broadcast()创建的广播变量在Executor端是只读的，但如果广播的对象太大，传输和反序列化的开销会很高。我一般只广播小型的配置表或字典，超过几十MB的数据宁可走分布式缓存或者重新读取。广播变量的spark.broadcast.blockSize默认4MB，调大到8MB或16MB可以减少传输次数，但会增加单次传输的内存压力。
5. 数据倾斜与分区异常：最考验经验的战场
5.1 怎么判断是不是数据倾斜
数据倾斜的典型表现是：大部分Task很快跑完，少数几个Task跑得特别慢，甚至卡在99%不动。在Spark UI的Stage页面里，看Task的Duration分布，如果最大值和75分位值差了几倍甚至几十倍，基本可以确定是倾斜。
另一个判断依据是Shuffle Read Size。如果某个Task的Shuffle Read Size远大于其他Task，那这个Task对应的分区就是倾斜分区。我一般会先跑一个df.groupBy("key").count().orderBy(desc("count")).show(20)，看看有没有某个key的数量级明显异常。
5.2 两阶段聚合：通用但要注意代价
两阶段聚合是解决倾斜的经典手段：先给key加随机前缀做一次局部聚合，再去掉前缀做全局聚合。这个方法的优点是通用，不依赖具体业务逻辑；缺点是增加了一次shuffle，整体耗时可能反而变长。
我一般只在倾斜非常严重（比如某个key占了总数据量的30%以上）时才用两阶段聚合。如果倾斜没那么严重，优先考虑调大分区数或者用repartition重新分布。分区数的经验值是：每个分区处理128MB到256MB数据。比如10GB的数据，分区数设在40到80之间比较合适。太小容易OOM，太大则Task调度开销增加。
5.3 空key和异常key的处理
空key（null或空字符串）是数据倾斜的常见来源。如果业务上不需要空key，直接过滤掉最省事。如果业务上需要保留，可以给空key加上随机后缀打散，最后再合并回来。
异常key指的是那些数量级明显偏离正常范围的key。处理方式取决于业务：如果是脏数据，过滤；如果是正常但集中的数据，加盐或单独处理。我遇到过一个模拟项目，某个key的数量是其他key的1000倍，最后是把这部分数据单独抽出来用广播变量处理，剩下的走正常流程，效果比硬扛好得多。
| 倾斜类型 | 推荐处理方式 | 代价 |
|---|---|---|
| 空key | 过滤或加随机后缀 | 低 |
| 少量大key | 加盐两阶段聚合 | 中 |
| 大量中等key | 调大分区数 | 低 |
| 极端大key | 单独抽出处理 | 高 |
6. 资源与调度异常：集群层面的坑
6.1 Executor Lost与动态资源分配
ExecutorLostFailure这个异常在开启动态资源分配（spark.dynamicAllocation.enabled=true）时特别常见。原因是Executor空闲超时被回收，但上面的shuffle数据还没被下游拉取。开了外部shuffle服务（spark.shuffle.service.enabled=true）之后，shuffle数据由外部服务管理，Executor回收不影响数据，这个问题就解决了。
动态资源分配的另一个坑是spark.dynamicAllocation.minExecutors设得太小，任务高峰期申请不到足够的Executor，导致Task排队。我一般会把minExecutors设成预估峰值的30%到50%，maxExecutors设成峰值的1.5倍，initialExecutors设成min和max的中间值。这样既能快速响应，又不会一直占着资源。
6.2 推测执行：用好了是利器，用不好是灾难
spark.speculation=true可以让Spark在某个Task跑得明显慢于其他Task时，启动一个备份Task。这个机制对解决倾斜导致的个别Task拖尾很有效，但副作用是可能重复执行有副作用的操作。如果你的Task里有写数据库、发消息这类操作，推测执行可能导致数据重复。
我一般只在纯计算任务里开推测执行，并且把spark.speculation.quantile（默认0.75）和spark.speculation.multiplier（默认1.5）调得保守一点，比如quantile设0.9，multiplier设2.0，减少不必要的备份Task。
6.3 队列资源与优先级
在YARN或Kubernetes这类资源调度框架上跑Spark，队列资源不足是任务卡住的常见原因。表现是Application一直处于ACCEPTED状态，不进入RUNNING。这种情况要去调度框架的界面看队列资源使用情况，而不是在Spark层面调参。
如果队列资源紧张，可以考虑用spark.yarn.queue指定一个相对空闲的队列，或者调整任务的优先级。有些调度框架支持抢占，但抢占会导致低优先级任务被kill，用之前要评估好影响。
7. 常见问题速查与避坑清单
7.1 异常速查表
| 异常关键字 | 最可能原因 | 第一优先动作 |
|---|---|---|
| Java heap space（Driver） | collect/take数据过多 | 改用foreachPartition或写回存储 |
| Java heap space（Executor） | 分区数据过大或缓存过多 | 调分区数、降缓存级别 |
| Direct buffer memory | 堆外内存不足 | 调大offHeap.size和memoryOverhead |
| FetchFailedException | Map输出丢失或拉取超时 | 查Executor稳定性、调重试参数 |
| Task not serializable | 闭包引用不可序列化对象 | 用mapPartitions重建资源 |
| KryoException | 未注册类或buffer不足 | 注册类、调大buffer.max |
| ExecutorLostFailure | 动态资源回收或节点故障 | 开外部shuffle服务、调超时 |
| Container killed by YARN | 物理内存超限 | 调大memoryOverhead |
7.2 我踩过的几个典型坑
第一个坑是盲目调大executor内存。早期遇到OOM就加内存，结果Executor内存加到32GB，GC停顿反而更长，任务更慢。后来才明白，单个Executor内存过大时，Full GC的停顿时间会显著增加，反而拖累整体性能。现在我的经验是单个Executor内存不超过16GB，超过就拆成多个小Executor。
第二个坑是忽略磁盘IO。有一次任务一直报shuffle写失败，查了半天内存和网络都没问题，最后发现是spark.local.dir指向的磁盘满了。Spark的shuffle文件、spill文件都写在那块盘上，盘满了自然写不进去。现在我会在任务启动前检查磁盘剩余空间，至少留出数据量1.5倍的空间。
第三个坑是参数照搬网上配置。网上很多“最优配置”是针对特定场景的，直接照搬到自己的任务上可能适得其反。比如spark.sql.shuffle.partitions默认200，有人建议调到2000，但如果你的数据量只有几GB，2000个分区会导致大量小文件和小Task，调度开销反而更大。参数一定要结合自己的数据量和集群规模来定。
7.3 排查问题的通用思路
遇到异常时，我的排查顺序一般是：先看Spark UI的Stage和Task页面，定位是哪个阶段、哪个Task出的问题；再看Executor的日志，找具体的异常堆栈；然后看YARN或Kubernetes的日志，确认是不是资源层面的问题；最后才去调参数。
这个顺序很重要，因为很多异常的表现和根因不在同一层。比如FetchFailedException表现是拉取失败，根因可能是Map端Executor被kill。如果一上来就调网络参数，可能永远解决不了问题。
提示：Spark UI的Environment页面会列出所有生效的配置，排查时先确认自己改的参数有没有真正生效。我见过好几次改了配置文件但任务读的是另一份配置的情况。
8. 一些关于调优节奏的个人体会
Spark性能优化这件事，最忌讳的就是“一把梭”。我刚开始做调优的时候，喜欢一次性改十几个参数，结果任务跑快了也不知道是哪个参数起了作用，跑慢了也不知道是哪个参数拖了后腿。后来学乖了，每次只改一到两个参数，跑完对比效果，确认有效再改下一个。虽然慢一点，但每一步都心里有数。
另一个体会是，大部分性能问题其实不是参数问题，而是数据问题和代码问题。数据倾斜、分区不合理、闭包引用过大对象，这些靠调参只能缓解，不能根治。真正有效的优化往往在代码层面：换一种写法、换一个算子、换一种数据结构，效果比调参明显得多。
最后说一个容易被忽略的点：监控和日志。任务跑的时候多看看Spark UI的实时指标，GC时间、shuffle读写量、Task分布，这些信息比事后看日志有用得多。我现在养成了习惯，任务启动后先盯几分钟UI，确认没有明显的倾斜和GC异常再去做别的事。这个习惯帮我提前发现过好几次潜在问题，省下了不少重跑的时间。
调优没有银弹，每个任务都有自己的脾气。多跑、多看、多记录，慢慢就能形成自己的判断体系。上面这些内容是我在实际操作中积累的一些经验，不一定适用于所有场景，但希望能给你提供一个排查问题的思路框架。
