3. 计算引擎调优：Spark SQL执行计划解析、Shuffle优化、动态资源分配与并行度设置

Spark 跑得慢，十有八九是这三个地方没调好：执行计划不合理、Shuffle 拖后腿、资源分配没跟上。我这些年踩过的坑，大部分都跟它们有关。

说白了，Spark 的优化就是三板斧——看懂计划、减少数据倾斜、让资源动起来。咱们一个一个说。

3.1 看懂 Spark SQL 执行计划

很多人遇到 SQL 慢，第一反应是加资源。其实很多时候，加资源不如改 SQL。但改之前，你得先看懂 Spark 到底是怎么跑的。

我个人习惯，遇到慢查询第一件事就是看执行计划。用 .explain("extended") 或者直接在 Spark UI 的 SQL 标签页里看。

3.1.1 执行计划的四个阶段

Spark SQL 的执行计划分四层，我整理了个表格，方便对照：

阶段 说明 常见问题
Parsed Logical Plan SQL 解析后的逻辑树，还没做优化 语法错误、表不存在
Analyzed Logical Plan 经过元数据校验，字段类型都确定了 类型不匹配、隐式转换
Optimized Logical Plan Catalyst 优化器做了谓词下推、列剪枝等 优化没生效，比如过滤条件没下推
Physical Plan 最终执行的物理计划，能看到 Shuffle 和 Join 策略 SortMergeJoin 代替了 BroadcastJoin

我最常看的是 Physical Plan。它直接告诉你 Spark 打算怎么干——用哪种 Join、有没有 Shuffle、数据怎么分区。

3.1.2 看懂 Physical Plan 的关键信息

举个例子，你执行 df.explain("extended")，输出里会有类似这样的内容：

== Physical Plan ==
*(5) SortMergeJoin [dept_id#10], [id#0], Inner
:- *(3) Sort [dept_id#10 ASC NULLS FIRST], false, 0
:  +- Exchange hashpartitioning(dept_id#10, 200)
:     +- *(2) Filter isnotnull(dept_id#10)
:        +- Scan parquet ...
+- *(4) Sort [id#0 ASC NULLS FIRST], false, 0
   +- Exchange hashpartitioning(id#0, 200)
      +- *(1) Filter isnotnull(id#0)
         +- Scan parquet ...

看到 SortMergeJoin 和 Exchange 了吗？这就是 Shuffle 的信号。如果两个表都不大，其实应该走 BroadcastHashJoin。为什么没走？大概率是表大小超过了 spark.sql.autoBroadcastJoinThreshold 的默认值 10MB。

我的习惯： 遇到 SortMergeJoin，先看能不能改成 BroadcastJoin。要么调大阈值，要么加 hint——SELECT /*+ BROADCAST(t1) */ ...。我在一个 5 亿行的大表关联小表场景里，光这一步就把耗时从 12 分钟降到了 40 秒。

3.2 Shuffle 优化——性能瓶颈的重灾区

Shuffle 是 Spark 最贵的操作，没有之一。数据要在不同节点间重新分区，涉及磁盘 I/O、网络传输、序列化。我见过太多任务，90% 的时间都耗在 Shuffle 上。

3.2.1 Shuffle 是怎么发生的？

触发 Shuffle 的操作主要有这些：

  • groupBy / reduceByKey / aggregateByKey——按 Key 聚合
  • join——除了 BroadcastJoin 以外的 Join 方式
  • distinct / repartition / coalesce——重新分区
  • orderBy / sort——全局排序

嗯，这里要注意：不是所有 Shuffle 都能避免，但我们可以让它不那么疼。

3.2.2 数据倾斜——Shuffle 的头号杀手

数据倾斜的表现很典型：大部分 Task 几秒跑完，就剩一两个 Task 卡在那，跑几个小时都不结束。

我曾经遇到过一个日志分析任务，按用户 ID 聚合。结果发现有一个超级用户，占了全量数据的 40%。那个 Task 直接 OOM 了。

解决数据倾斜，我常用的几个方法：

  1. 加盐（Salting）——给热点 Key 加随机前缀，打散到多个分区
  2. 调整并行度——增大 spark.sql.shuffle.partitions，让每个分区数据更均匀
  3. 两阶段聚合——先局部聚合，再全局聚合

拿加盐来说，代码大概长这样：

// 给热点 Key 加随机前缀
val saltedDf = df.withColumn("salted_key", 
  when(col("key") === "hot_key", concat(col("key"), lit("_"), (rand() * 10).cast("int")))
  .otherwise(col("key")))

// 按加盐后的 Key 聚合
val result = saltedDf.groupBy("salted_key").agg(sum("value") as "partial_sum")
  // 去掉前缀，再聚合一次
  .withColumn("original_key", regexp_replace(col("salted_key"), "_\\d+$", ""))
  .groupBy("original_key").agg(sum("partial_sum") as "total_sum")
避坑： 加盐的随机范围不要太小。我一开始只分了 5 个桶，结果还是倾斜。后来改成 100 个，效果立竿见影。另外，加盐只适用于聚合场景，Join 场景要用其他方案。

3.2.3 Shuffle 参数调优

几个关键参数，我列了个表：

参数 默认值 我的建议 说明
spark.sql.shuffle.partitions 200 根据数据量调整，通常 500-2000 分区数太少会导致单个分区数据过大
spark.shuffle.file.buffer 32k 64k-128k 增大缓冲区减少磁盘 I/O 次数
spark.reducer.maxSizeInFlight 48m 96m-128m 增大拉取缓冲区，减少网络请求次数
spark.shuffle.compress true 保持 true 压缩 Shuffle 中间数据，减少网络传输

我个人习惯，先看 Shuffle 的数据量。如果 Shuffle 读写了几个 TB，那分区数至少得 1000 起步。否则每个分区处理的数据太多，GC 压力会很大。

3.3 动态资源分配与并行度设置

很多团队喜欢固定 Executor 数量。但实际场景里，数据量是波动的。白天跑批任务多，晚上跑报表。固定资源要么浪费，要么不够用。

动态资源分配就是干这个的——让 Spark 根据负载自动调整 Executor 数量。

3.3.1 开启动态资源分配

配置很简单，核心就这几个：

spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=2
spark.dynamicAllocation.maxExecutors=100
spark.dynamicAllocation.initialExecutors=5
spark.dynamicAllocation.executorIdleTimeout=60s
spark.dynamicAllocation.schedulerBacklogTimeout=1s

逻辑是这样的：当有 Task 在排队等待超过 1 秒（schedulerBacklogTimeout），就申请新的 Executor。当 Executor 空闲超过 60 秒，就释放掉。

注意： 动态资源分配需要 YARN 或 Kubernetes 支持。如果你用的是 Standalone 模式，得配合 External Shuffle Service 才能正常工作。我刚开始用的时候没配这个，结果 Executor 释放后 Shuffle 数据丢了，任务全挂。

3.3.2 并行度怎么设？

并行度是个老生常谈的问题。设小了，资源利用率低；设大了，调度开销大。

我的经验公式：

  • 每个 Executor 的 Core 数 × Executor 总数 × 3 到 5 倍，作为 Task 并行度
  • 比如 10 个 Executor，每个 4 Core，那并行度在 120 到 200 之间比较合理

具体到参数：

  • spark.sql.shuffle.partitions——控制 Shuffle 后的分区数
  • spark.default.parallelism——控制 RDD 操作的默认并行度
  • df.repartition(n) / df.coalesce(n)——手动调整分区

我一般先看每个分区的数据量。理想情况是每个分区 100MB 到 200MB。如果分区数据量超过 1GB，那并行度肯定不够。

3.3.3 动态资源 + 并行度的配合

这里有个容易忽略的点：动态资源分配和并行度是联动的。

假设你设了 spark.sql.shuffle.partitions=200，但动态资源只分配了 5 个 Executor，每个 2 Core。那总共只有 10 个 Slot。200 个 Task 要分 20 批才能跑完。调度开销会很大。

反过来，如果你设了 2000 个分区，但只有 10 个 Executor，那每个 Executor 要处理 200 个 Task。虽然能跑，但效率不高。

我的做法是：

  1. 先根据历史数据估算峰值资源，设一个合理的 maxExecutors
  2. 并行度按 maxExecutors × coresPerExecutor × 3 来设
  3. 如果任务波动大，可以适当提高并行度，让动态资源去适配

3.4 本章知识体系

下面这张图把三个核心模块串起来了。你可以看到，执行计划是诊断入口，Shuffle 优化是核心战场，动态资源是弹性保障。三者缺一不可。

Spark SQL 性能优化知识体系 Spark SQL 性能优化三大核心模块 执行计划解析 Parsed Logical Plan Analyzed Logical Plan Optimized Logical Plan Physical Plan 诊断入口：explain("extended") Shuffle 优化 数据倾斜处理 加盐 / 两阶段聚合 Shuffle 参数调优 BroadcastJoin 替代 核心战场：减少数据移动 动态资源分配 min/maxExecutors Executor 动态伸缩 并行度联动设置 External Shuffle Service 弹性保障：按需分配 优化流程：先看执行计划 → 定位 Shuffle 瓶颈 → 调整资源与并行度 三者联动，缺一不可。只看计划不改 Shuffle 是纸上谈兵，只加资源不看计划是浪费。 核心参数速查 spark.sql.shuffle.partitions = 500-2000 spark.dynamicAllocation.maxExecutors = 100 spark.shuffle.file.buffer = 64k-128k spark.sql.autoBroadcastJoinThreshold = 50m spark.reducer.maxSizeInFlight = 96m-128m spark.dynamicAllocation.executorIdleTimeout = 60s

3.5 实战建议

最后，给几个落地建议：

  • 先诊断，后调优。 别一上来就改参数。先看 Spark UI 的 SQL 页面，找到最慢的 Stage，看它是不是 Shuffle 引起的。
  • 一次只改一个参数。 我见过有人一次改 10 个参数，结果性能变差了，都不知道是哪个改坏的。
  • 用历史数据做基准。 每次调优前记录一下基线——数据量、耗时、Shuffle 读写量。调完对比，才知道效果。
  • 动态资源不是银弹。 如果任务本身有严重的数据倾斜，加再多 Executor 也没用。先把倾斜解决了，再谈弹性伸缩。

嗯，这一章的内容就到这。执行计划、Shuffle、动态资源，这三个点吃透了，Spark 性能优化你就掌握了 80%。剩下的 20%，是在实战中慢慢积累的。

← 上一章 📖 返回目录 下一章 →