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.registrationRequiredfalse让Kryo自动处理但这样会失去一部分性能优势。折中方案是只注册那些频繁序列化的核心类比如自定义的POJO和枚举。// 自定义Kryo注册器示例 class MyKryoRegistrator extends KryoRegistrator { override def registerClasses(kryo: Kryo): Unit { kryo.register(classOf[MyCustomRecord]) kryo.register(classOf[MyEnum]) } } // 配置spark.kryo.registratorcom.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的处理空keynull或空字符串是数据倾斜的常见来源。如果业务上不需要空key直接过滤掉最省事。如果业务上需要保留可以给空key加上随机后缀打散最后再合并回来。异常key指的是那些数量级明显偏离正常范围的key。处理方式取决于业务如果是脏数据过滤如果是正常但集中的数据加盐或单独处理。我遇到过一个模拟项目某个key的数量是其他key的1000倍最后是把这部分数据单独抽出来用广播变量处理剩下的走正常流程效果比硬扛好得多。倾斜类型推荐处理方式代价空key过滤或加随机后缀低少量大key加盐两阶段聚合中大量中等key调大分区数低极端大key单独抽出处理高6. 资源与调度异常集群层面的坑6.1 Executor Lost与动态资源分配ExecutorLostFailure这个异常在开启动态资源分配spark.dynamicAllocation.enabledtrue时特别常见。原因是Executor空闲超时被回收但上面的shuffle数据还没被下游拉取。开了外部shuffle服务spark.shuffle.service.enabledtrue之后shuffle数据由外部服务管理Executor回收不影响数据这个问题就解决了。动态资源分配的另一个坑是spark.dynamicAllocation.minExecutors设得太小任务高峰期申请不到足够的Executor导致Task排队。我一般会把minExecutors设成预估峰值的30%到50%maxExecutors设成峰值的1.5倍initialExecutors设成min和max的中间值。这样既能快速响应又不会一直占着资源。6.2 推测执行用好了是利器用不好是灾难spark.speculationtrue可以让Spark在某个Task跑得明显慢于其他Task时启动一个备份Task。这个机制对解决倾斜导致的个别Task拖尾很有效但副作用是可能重复执行有副作用的操作。如果你的Task里有写数据库、发消息这类操作推测执行可能导致数据重复。我一般只在纯计算任务里开推测执行并且把spark.speculation.quantile默认0.75和spark.speculation.multiplier默认1.5调得保守一点比如quantile设0.9multiplier设2.0减少不必要的备份Task。6.3 队列资源与优先级在YARN或Kubernetes这类资源调度框架上跑Spark队列资源不足是任务卡住的常见原因。表现是Application一直处于ACCEPTED状态不进入RUNNING。这种情况要去调度框架的界面看队列资源使用情况而不是在Spark层面调参。如果队列资源紧张可以考虑用spark.yarn.queue指定一个相对空闲的队列或者调整任务的优先级。有些调度框架支持抢占但抢占会导致低优先级任务被kill用之前要评估好影响。7. 常见问题速查与避坑清单7.1 异常速查表异常关键字最可能原因第一优先动作Java heap spaceDrivercollect/take数据过多改用foreachPartition或写回存储Java heap spaceExecutor分区数据过大或缓存过多调分区数、降缓存级别Direct buffer memory堆外内存不足调大offHeap.size和memoryOverheadFetchFailedExceptionMap输出丢失或拉取超时查Executor稳定性、调重试参数Task not serializable闭包引用不可序列化对象用mapPartitions重建资源KryoException未注册类或buffer不足注册类、调大buffer.maxExecutorLostFailure动态资源回收或节点故障开外部shuffle服务、调超时Container killed by YARN物理内存超限调大memoryOverhead7.2 我踩过的几个典型坑第一个坑是盲目调大executor内存。早期遇到OOM就加内存结果Executor内存加到32GBGC停顿反而更长任务更慢。后来才明白单个Executor内存过大时Full GC的停顿时间会显著增加反而拖累整体性能。现在我的经验是单个Executor内存不超过16GB超过就拆成多个小Executor。第二个坑是忽略磁盘IO。有一次任务一直报shuffle写失败查了半天内存和网络都没问题最后发现是spark.local.dir指向的磁盘满了。Spark的shuffle文件、spill文件都写在那块盘上盘满了自然写不进去。现在我会在任务启动前检查磁盘剩余空间至少留出数据量1.5倍的空间。第三个坑是参数照搬网上配置。网上很多“最优配置”是针对特定场景的直接照搬到自己的任务上可能适得其反。比如spark.sql.shuffle.partitions默认200有人建议调到2000但如果你的数据量只有几GB2000个分区会导致大量小文件和小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异常再去做别的事。这个习惯帮我提前发现过好几次潜在问题省下了不少重跑的时间。调优没有银弹每个任务都有自己的脾气。多跑、多看、多记录慢慢就能形成自己的判断体系。上面这些内容是我在实际操作中积累的一些经验不一定适用于所有场景但希望能给你提供一个排查问题的思路框架。