接手过不少 Spark 任务我见过最多的问题不是集群资源不够而是代码本身写得太随意。同样的数据量一段规范代码可能 20 分钟跑完换成一段只图“能出结果”的代码拖到 2 小时还卡在某个 Stage 上。所谓 Spark 代码规范并不是为了好看而是在约束每一行代码对数据移动、内存占用、执行计划的影响。这篇博文围绕高性能 Spark 应用实践把我在生产环境里沉淀下来的写代码习惯、参数配置、排查思路一起过一遍适合正在写业务数仓任务、实时清洗逻辑或者算法特征加工的同学参考。1. 先把性能问题的账算明白为什么坏代码比小集群更致命1.1 一个真实的对照两段代码两种命运我在日常运维中做过一次对照。两份逻辑几乎一样的任务都是处理 300GB 左右的用户行为日志一份由某个临时脚本改造过来另一份按项目规范重写。前者的做法是读 JSON 后用 RDD 的 map 逐条解析再 filter 掉无效字段最后 groupByKey 做聚合后者直接读成 DataFrame提前过滤、只留需要的列用 reduceByKey 或者 DataFrame API 完成聚合。两边都跑在同一个队列上资源配额一模一样。结果很典型规范重写后的任务整个 DAG 只有 6 个 StageShuffle 数据量大约 40GB原始脚本跑了 14 个 StageShuffle 数据量超过 120GB中间还有两次因为 Executor OOM 导致整个 Application 失败重试。最后耗时一个 18 分钟一个 1 小时 50 分钟。这个差异完全不靠调参纯粹是代码路径不同。这就是我想强调的观点性能优化不是等任务慢了才去翻 Spark UI而是在写第一行代码时就考虑这条数据会被切成几份、哪些列会跨节点传输、聚合在哪一端发生。否则后面无论怎么加 Executor、调并行度都是在给一个坏结构打补丁。1.2 规范的本质尽可能减少“数据搬家”Spark 计算模型里最贵的操作不是计算本身而是数据在网络和磁盘之间的移动。这个概念我用生活里的事打比方几个人一起做饭菜在同一个厨房时切菜、炒菜、装盘都很快但如果要求每道菜都先送到另一个城市去加工再运回来整个过程就会变得又慢又不可控。Spark 里的 Shuffle 就是这个“异地加工”。具体来说Shuffle 产生的原因主要是两类一是按 Key 重新分区比如 groupByKey、reduceByKey、distinct、join二是通过 repartition 改变分区数量。每一次 Shuffle 都意味着父 Stage 的所有输出要先写磁盘再被网络拉取到下一批 Task 里。磁盘 IO、网络 IO、序列化开销全在这里爆发。所以代码规范的第一条原则可以用一句话概括任何逻辑表达都要先问一句这一步会不会引起全量数据重新洗牌能不能在 Shuffle 之前把数据量砍掉一截。比如过滤条件能提前就提前列能少选就少选需要聚合的 Key 尽量在 map 端做一次预聚合。这不是“洁癖”是在控制成本。2. 数据抽象选错了后面所有优化都是白做2.1 DataFrame 优先但 RDD 什么时候还要用很多从老项目接手代码的人第一反应就是 RDD 很灵活什么都能干。确实RDD 暴露了分区、依赖、计算函数写起来很直接。但灵活的另一面是Spark 的查询优化器很难介入 RDD 代码它不知道你的 map 函数里到底读了哪些字段、哪些数据会被过滤掉因此很多 RDD 版本的逻辑连最基本的“列剪枝”都享受不到。我现在的默认原则是能写 DataFrame / SQL 就绝不写 RDD。DataFrame 有 schemaCatalyst 优化器能看清楚数据的形状能做谓词下推、列剪枝、常量折叠甚至能在 join 时自动选择 Broadcast 还是 SortMerge。这份红利是白拿的。RDD 也不是完全没有用。碰到三类场景我会退回 RDD一是数据本身完全无结构比如二进制报文、带自定义分隔符的杂散文本强行套 schema 反而别扭二是需要自己控制每个分区的处理粒度比如某些模型打分要保证一批样本在同一个 Task 内完成三是需要精准控制分区器的 hash 规则避免默认分区把热点 Key 打散成奇怪的分布。除此之外能用 DataFrame 就 DataFrame。2.2 Catalyst 优化器到底替你省了哪几步Catalyst 优化器做的事情简单拆解有四步逻辑计划解析、逻辑计划优化、物理计划生成、代码生成。其中逻辑优化阶段的谓词下推和列剪枝对性能的影响最直白。举个例子。有一张订单表和一张用户表做 join最后只要用户维度的等级字段和订单金额。如果你先 join 再 filter 用户状态优化器会自动尝试把 filter 下推到 scan 阶段让每个分区只读需要的行。如果你在 DataFrame 上做了 select只保留三列那么 scan 时就不会再去解析和传输剩下几十个字段。这些优化在 RDD 版本里都不会发生。但要注意Catalyst 不是万能的。有些优化受限于表达方式。比如在 DataFrame API 里写一个自定义 UDF优化器就看不清 UDF 内部逻辑既不能下推也不能剪枝整列数据都会被拉进函数里做序列化和反序列化。能用内置函数解决的问题尽量别造自定义 UDF这是 DataFrame 时代最容易踩的隐性坑。2.3 一个 join 写法的执行计划对比我实际演示过两种 join 写法。数据是两张分布表一张 2 亿行一张 3 万行的小表。第一种写法直接val result bigDF.join(smallDF, Seq(key))没加任何提示Spark 默认通过自动 Broadcast Join 阈值处理执行计划里出现 BroadcastHashJoin没有 Shuffle。第二种写法是同一个 join但有人在 join 前对两张表做了 repartition(2000)这一下就把 Broadcast 计划打破了物理计划变成 SortMergeJoin额外增加了一次全量 Shuffle跑了 8 分钟。而第一种只要 30 秒。这个案例说明很多性能问题不是“不会优化”而是无意间破坏了优化器的可优化条件。频繁 repartition、过多的 select 嵌套、把数据写成中间表再读回来都会让优化器束手束脚。写完一段 SQL 或 DataFrame 链式操作后应该养成一个习惯explain() 看一眼物理计划确认 join 是 Broadcast 还是 SortMergefilter 是不是真的下推到了 scan。这几秒钟的检查比在集群上跑完一遍再猜快得多。3. 逐行拆解 Spark 代码规范从分区、join 到算子选型3.1 分区设计repartition 与 coalesce 怎么选分区数量直接决定 Task 数量和并行度。太大的分区数会导致调度开销暴涨整个集群出现大量只有几 MB 输出的空 Task太小则浪费并行能力单 Task 处理时间过长。我比较常用的估算公式是目标分区数 ≈ 总输入数据量 / 期望单 Task 处理数据量。日志类清洗任务里单 Task 处理 128MB 到 256MB 比较合理复杂聚合可以放宽到 500MB 左右。repartition 和 coalesce 的差别很多人只记住了“一个会引起 Shuffle一个不会”但没有深挖。coalesce 只能减少分区而且是尽量把同一节点上的分区合并天然能减少数据跨节点传输repartition 可以增加也可以减少分区shuffle 一定会发生。我在写最终结果到 Hive 表时几乎只用 coalesce 控制输出文件数量而不会在前面随意 repartition。如果并行度不够需要提高分区数那只能用 repartition这是没办法绕开的代价。有一个容易踩的坑coalesce 之后紧跟一个 map 类算子很可能会打破 coalesce 的窄依赖结构导致 Spark 重新做一次 Shuffle。这个现象在 Spark UI 里会看到明明写了 coalesceStage 数量却比你预期多。真要控制最终输出文件数应该把 coalesce 放在所有转换的最后一步或者直接在 write 前用 coalesce不要在它后面再接复杂的 map。3.2 join 策略小表广播不是唯一招数广播变量是最常见的优化手段之一。把几十 MB 的维度表做成广播可以让每个 Executor 在本地保留一份完整数据join 时完全不走 Shuffle。阈值参数是 spark.sql.autoBroadcastJoinThreshold默认 10MB生产上我建议根据 Executor 内存适量调大到 20MB 或 50MB但别无限调大否则每个 Executor 的堆内都要存一份内存反而吃紧。手动指定广播也有技巧不要只依赖自动阈值。大表 join 小表时可以在 SQL 里写SELECT /* BROADCAST(dim) */ * FROM fact f JOIN dim d ON f.key d.keyDataFrame API 则用factDF.join(broadcast(dimDF), key)除了 join广播变量还广泛用在“遍历维度做映射”的场景。比如一个 10 万条目的大字典如果放在 mapPartitions 里逐条查询造成重复序列化性能会很差。正确做法是用 SparkContext.broadcast 把字典打到每个 Executor然后在 UDF 或 mapPartitions 里直接引用 broadcast .value省掉每一次 shuffle 和网络查询。另一个思路是不用广播而是做 Bucket Join。如果两张表都按同一个 Key 预先分桶成相同数量的文件join 时可以做 Bucket 局部合并连 Shuffle 都省掉。代价是建表和写入时需要稳定的分桶操作属于数仓建模阶段就要设计的方案不适合在临时任务里临时抱佛脚。3.3 算子选型groupByKey 是反面教材mapPartitions 是宝藏RDD 算子选型是最能拉开代码差距的地方。groupByKey 为什么是反面教材因为它的逻辑是“把所有相同 Key 的 value 全量洗牌到一个分区再统一处理”。哪怕你已经知道每个 Key 的求和值非常小它也会把每个原始 record 全部搬一遍。reduceByKey 则完全不同它在 map 端先对每个分区内部的相同 Key 做一次预聚合Shuffle 时传输的已经是中间聚合结果。一个计算 PV/UV 的任务原始数据 100GB预聚合后可能只有 1GBShuffle 量直接差两个数量级。代码上只是从 groupByKey().mapValues(.sum) 改成 reduceByKey( _)执行计划却完全两样。另一个高频场景是“对每个分区做批量初始化”。比如每台 Executor 上要加载一个模型文件、建立数据库连接很多人会用 map 每条记录初始化一次连接数被放大到百万级。正确做法是 mapPartitions让每个 Task 只初始化一次再在循环里复用。写数据库时同理foreachPartition 配合连接池比每条记录 foreach 里创建连接稳定得多。我见过比较无语的代码是 map 里创建了 SimpleDateFormat每条数据都 new 一次对象导致 GC 压力巨大。这类对象应该放到 mapPartitions 外面统一初始化或者直接用线程安全的 DateTimeFormatter 常量。一个看似无关紧要的算子选择最后可能换来几十分钟的 GC 耗时。3.4 序列化与内存两个最容易忽略的配置项Spark 默认的 Java 序列化器虽然稳定但生成的字节数组特别膨胀序列化速度也慢。Kryo 序列化在压缩率和速度上都明显占优不过它需要提前注册类否则也会退化到不可控状态。我的日常配置是spark.serializer org.apache.spark.serializer.KryoSerializer spark.kryo.registrationRequired true spark.kryo.registrator com.example.MyRegistrator如果使用 DataFrame API很多序列化工作被 Catalyst 优化成内部二进制格式对这种配置的敏感度会低一些但涉及 RDD、UDF、广播变量或者自定义对象时Kryo 带来的收益非常可观。内存上要理解 Spark 的 Unified Memory 模型。Executor 堆内内存被切成两块Execution 内存和 Storage 内存由 spark.memory.fraction 控制可用比例默认 0.6另外 0.4 是保留区主要用于记录内部对象和防止 OOM。如果代码里频繁 cache 大表又同时跑大 shuffle两者会互相挤压。我一般建议不要盲目调大 spark.memory.fraction因为保留区太小会让 JVM 在内存分配上变得脆弱遇到极端情况反而更容易崩。真正容易忽略的是 spark.executor.memoryOverhead。它不属于堆内 Execution 内存但包含了 JVM 线程开销、NIO buffer、内部元数据。很多人的 OOM 问题不是堆内不够而是 overhead 没给够。默认值是 executor 内存的 10%碰上大规模 shuffle、大量 UDF 或者使用 Kryo 时我通常手动配置成 1GB 甚至 2GB。这个参数在 Spark UI 里会显示为 “Memory Overhead”被很多排查的人一眼略过。4. 参数调优与执行计划让规范落地到真实集群4.1 内存与并行度参数的计算套路调参先要有个估算过程不是抄一个生产环境的配置就完事。假设一个 Executor 有 8GB 堆内内存4 个核。通常我不会把每个 Executor 的内存拉满而是留出 spark.executor.memoryOverhead 的空间。实际可用内存可以按堆内 × spark.memory.fraction 估算也就是 8GB × 0.6 ≈ 4.8GB 的 Unified 区域。这个区域内既要放 shuffle 缓存又要放 cache 数据还要运行聚合的 HashMap。并行度设置可以参考总核心数。比如集群有 20 个 Executor每个 4 核总并行 Task 数就是 80。那么 Spark 中一个 Stage 的理想 Task 数量最好在 80 到 160 之间太少会浪费空闲核太多则每个 Task 处理数据量过小、调度成本变大。尤其要注意 spark.sql.shuffle.partitions默认值 200很多任务在数据量只有几百 MB 时也强行切成 200 个分区。此时每个分区只有几 MB网络和序列化开销占比非常高。我习惯根据数据量和并行度重新设它一般不超过集群并发核数的 2 到 3 倍。参数并不是越多越好。我见过有人把几十个 Spark 参数写进同一套提交脚本然后发现照样慢。原因是很多参数是互相耦合的比如 Executor 内存加大后默认并行度没变每个 Task 能吃到的内存更多但 CPU 核数没增加整体吞吐量不会有质变。做参数调整最好一次只改一个变量用 Spark UI 里各个 Stage 的耗时来做前后对比。4.2 动态分配与 AQE自动优化能救多少代码问题Spark 3.0 之后的 Adaptive Query ExecutionAQE是我现在默认开启的东西。它能在运行时根据已经完成的 Stage 统计自动调整后续的 Shuffle 分区数、自动转换 SortMergeJoin 为 BroadcastHashJoin、自动处理 Join 时的数据倾斜。配置如下spark.sql.adaptive.enabled true spark.sql.adaptive.coalescePartitions.enabled true spark.sql.adaptive.skewJoin.enabled trueAQE 解决了我之前手动调参的一半工作。以前要反复试 200、400、800 个 Shuffle 分区现在它能根据 map 端的输出大小自动计算合适的 reduce 分区数。以前遇到大表 join 小表偶尔因为阈值判断问题走了 SortMergeJoin现在它能在 shuffle 过程中发现小表足够小动态改成 Broadcast 策略。但不要认为开启 AQE 就可以对代码完全放飞。AQE 主要优化的是 Shuffle 之后的 Reduce 端它没办法弥补 map 端被 filter 过滤掉的行数、无法消除用户写出来的重度 UDF、也无法替代“尽量少读无用列”的自觉。它更像是安全网可以兜底一部分不合理的并行度选择但不能替代规范本身。4.3 从 Spark UI 复盘一段慢任务的完整思路慢任务排查时我通常遵循一条固定路线先看 Job 列表里哪些 Stage 耗时最长认准大 Shuffle 发生的位置然后点进 Stage 看 Shuffle Read Size / Shuffle Write Size如果 Write 比输入还大说明在算子选择上产生了大量中间数据最后看 Event Timeline 里的 Executor 分布判断是不是数据倾斜导致少数 Task 特别慢。有一次我定位一个“波浪形”运行的任务从 UI 上看到某个 Stage 的 200 个 Task 里199 个只用了 5 秒剩下 1 个跑了 40 分钟Shuffle Read 接近全量数据。这种一眼就能断定是 Key 倾斜。如果代码里用 DataFrame API可以通过 explain 看到实际执行计划再针对倾斜 Key 做处理。我还特别喜欢看 GC Time。某个 Stage 的 GC 时间超过总运行时间 20% 时基本可以断定代码创建了过多短生命周期对象。此时不一定要调大内存而应该检查是不是在 map 里频繁造对象、用了不合理的 UDF、或者把大数据集 cache 到了 Storage 区域导致 Execution 内存被挤压。UI 上的指标不是用来截图汇报的是用来做推理的证据。5. 生产环境常见问题与排查技巧实录5.1 归因 OOM先弄清楚是谁的内存OOM 可以说是 Spark 任务里最常见的失败原因。但“OOM”只是一个结果归因错误就会白调参。我会先看日志里的异常类如果是 java.lang.OutOfMemoryError: Java heap space说明堆内内存不足如果是 Unable to acquire X bytes of memory多半是堆外内存或 Off-heap 不足如果是 Direct buffer memory通常是 NIO 或 Kryo 的堆外缓冲问题。堆内 OOM 常见于 reduce 端聚合过猛、collect() 拉取全量结果、cache 数据量超过 Storage 区域。很多人以为加大 spark.executor.memory 就行但如果问题的根源是代码里有个 groupByKey 把所有数据都拉到单个分区加内存只是延缓爆掉的时间。我的原则是OOM 出现时先去看 Stage 的 Shuffle Read 和记录数如果某 Task 拿到的是其他 Task 几百倍的输入优先处理倾斜如果整体都很大再考虑调内存。还有一个隐蔽问题driver 端 OOM。当代码里执行了 collect() 把几亿行数据收回到 driver或者大量使用 broadcast 变量列表driver 的默认 1GB 内存很容易炸。这种情况调 Executor 内存毫无意义。规范上要避免 collect 大数据集如果只是抽样观察数据用 take(100) 或者 limit 就够了。5.2 数据倾斜定位、拆分、两阶段聚合数据倾斜的经典场景是 join 和 groupBy。定位方法可以直接看 Spark UI某个 Stage 里 Task 耗时严重不均个别 Task 的 Input 或 Shuffle Read 远高于中位数。确认是某个 Key 倾斜后最直接的手段是两阶段聚合。两阶段聚合的思路是第一次为 Key 加上随机前缀打散到多个分区做部分聚合第二次去掉前缀再对真实 Key 做全局聚合。用 Scala 实现大致是rdd.map { case (key, value) val salt Random.nextInt(10) ((key, salt), value) }.reduceByKey(_ _) .map { case ((key, _), sum) (key, sum) } .reduceByKey(_ _)前缀范围不宜太大10 到 100 之间可以接受。太大反而会造成第一轮 Stage 压力过大。对于 join 倾斜可以先用 filter 把倾斜 Key 单独拆出来加上随机前缀后与右表膨胀副本 join再和正常 Key 的 join 结果合并。Spark 3 的 AQE 里已经内置了 skewJoin 优化但前提是开启后且满足阈值复杂场景仍然需要手动拆。这个方案的关键是理解成本两阶段聚合不会消除计算量只是把原本集中在同一个 Task 上的压力分散到多个 Task让每个 Task 都能吃到集群的并行能力。它不能解决单 Key 的总数据量超过单机内存的根本问题所以有时候还要配合过滤掉低频噪声数据。5.3 小文件问题写入端管不住读端就要哭小文件是一个非常隐蔽但持续影响性能的问题。Hive 表如果写入时分区内出现几百乃至上千个小文件后续读取时每个文件都会至少起一个 Task元数据开销、打开文件耗时、任务调度开销全被放大。这种问题往往不是优化一段代码就能修复它会拖累所有读取这张表的任务。写入端最常见的做法是写入前 coalesce 到合理分区数。比如最终要写入一个分区的 2GB 数据期望单文件 256MB那么目标文件数就是 8 个左右。Spark 写入时可以用df.coalesce(8).write.mode(overwrite).format(parquet).saveAsTable(xxx)如果是动态分区写入每个分区都可能有分布不均匀的数据直接把所有分区压到过少的分区又会产生大文件因此需要结合 AQE 的动态分区裁剪和写入端 coalesce 参数来处理。一般我会开启 spark.sql.adaptive.coalescePartitions.enabled并且检查每个分区的数据量是否与期望文件数匹配。测试环境里看起来不痛不痒的小文件问题生产环境累积一个月后会产生巨大的读放大。我也遇到过写入端压力不大但 Hive 表本身已存在大量小文件的存量问题。这时候只能做一次合并重写把分区里的小文件读出来再 coalesce 写回同等数据。这种操作既花时间又耗资源所以更应该在规范上堵住源头任何覆盖写表的任务提交前先算一遍目标文件数。5.4 一张问题排查速查表我整理了一份常用的排查对照表能覆盖大部分日常问题现象可能原因快速排查方式个别 Task 耗时远高于中位数Key 倾斜或 Executor 资源不均Spark UI 查看 Task Input / Shuffle Read 分布Stage 耗时高且 Shuffle Write 极大使用 groupByKey、过度 repartition查看物理计划改 reduceByKey 或减少 repartitionExecutor 堆内 OOM聚合过重、数据倾斜、collect 大结果先定位 Stage再判断是堆内还是堆外Driver OOMcollect / collectAsList 数据量过大搜 driver 日志改用 take 或 limit总 GC 时间过长map 内频繁创建对象、UDF 权重过大查看 GC Time 指标改用 mapPartitions 复用对象读取 Hive 表极慢小文件过多查看分区文件数执行 compaction 合并结果正确但运行时间波动动态资源未开启、并发干扰打开动态分配隔离队列稳定资源配置这张表不能替代经验判断但适合作为第一步的入口。排查问题时不建议一次改三个参数最好一次只改一个再跑一次看 UI 变化否则出了问题都不知道是哪个变量引起的。6. 这些规范落地时的个人体会我从这些年的 Spark 开发里得到的一条最重要的体会是高性能代码不是写出来的是“审”出来的。团队里定规范容易执行起来难。后来我们在每次代码评审时把重点从“结果对不对”扩展为“结果对之外还要看执行计划”每个人提交前都要贴一份 explain 关键输出评审的人先看有没有大 Shuffle、有没有能提前的过滤、有没有不该存在的 repartition。这样几轮下来新同学也能很快建立起对数据移动的直觉。还有一个细节想分享cache 慎用。很多同学习惯对中间结果 cache 一下以为能加速但如果没有复用cache 纯属白占内存甚至可能导致后续 Shuffle 的内存不足。判断标准很简单同一个 DataFrame 在 DAG 里被分支引用超过一次时才值得 cache否则不要随便加。最后说说“先优化代码还是先加资源”的取舍。我的建议是出现性能问题时先在现有资源下去看代码和 UI找到明显瓶颈后做针对性修改只有在确认代码已经相对规范、瓶颈来自数据量级本身时才去申请资源扩缩容。资源能解决一时的问题但规范不佳的任务明天会继续产生新的问题。把这几条原则带进日常迭代才是 Spark 应用长期稳定高效的基础。