Spark 原理与调优

本章目标：能定位 Spark 作业慢在哪，解释 shuffle 和数据倾斜，能写出加盐、两阶段聚合、广播 join，并会算 executor 和分区参数。 前置知识：读完 PySpark 实战；理解 Job/Stage/Task、宽窄依赖和 shuffle。 预计用时：2.5 小时。

1. 为什么需要它

同一段代码，10 GB 数据跑 2 分钟，1 TB 数据可能跑 2 小时，甚至 OOM。调优不是背参数，而是按顺序排除：

1. 先保证口径正确：结果不对，再快也没用。
2. 减少数据量：尽早过滤、只选需要的列、用分区裁剪。
3. 减少 shuffle：能广播就广播，能预聚合就预聚合。
4. 处理倾斜：让每个 Task 的数据量接近。
5. 调资源：executor 数量、内存、核数、分区数。
6. 调文件：避免小文件和过多分区。

先看症状：

2. 核心概念

2.1 Shuffle 的原理与开销

shuffle 是把数据按 key 重新分发的过程。以 groupBy("category") 为例：

Map 端：按 key 算 hash -> 分成 N 份 -> 排序、溢写、合并到本地磁盘

                         | 网络拉取
Reduce 端（N 个 Task）：拉取属于自己的份 -> 归并 -> 聚合 -> 写出

开销来自五处：磁盘 I/O（溢写时反复读写）、网络传输（跨机架更贵）、序列化/反序列化（CPU 消耗）、排序/归并（相同 key 要聚在一起）、内存压力（缓冲区、聚合状态、拉取数据同时占内存，容易 spill 或 OOM）。

shuffle 文件数约等于 Map Task 数 × Reduce 分区数。两边都很大时，小文件和随机读会拖垮磁盘。Spark SQL 默认 spark.sql.shuffle.partitions=200，这个值对 1 GB 和 1 TB 不可能都合适；AQE 会根据数据量合并分区，但不要完全依赖默认值。

2.2 数据倾斜：少数 Task 干了大部分活

现象：一个 Stage 里大部分 Task 几分钟跑完，个别 Task 跑一小时；Stage 长期停在 99%；某个 Executor 的 shuffle read 远大于其他；频繁 spill 甚至 OOM。

原因：某个 key 的数据量远大于其他。常见来源：

• 空值/未知值：customer_id 为空、province 为“未知”，所有空值落到同一个分区。
• 热点业务 key：头部商家、爆款商品、默认城市、测试账号。
• join key 分布不均：大表 join 大表，某个 key 在两边都很多。
• 分区函数：hash 后仍然集中，或分区数太少。
• 数据增长：某天活动导致某个品类订单暴涨。

检测：

key_counts = df.groupBy("key").count().orderBy(F.col("count").desc())
key_counts.show(10)
key_counts.agg(F.max("count").alias("max_cnt"), F.avg("count").alias("avg_cnt")).show()

max_cnt / avg_cnt 超过几十倍，基本可以怀疑倾斜。Spark UI 的 Stage 页面看 Summary Metrics：比较 Shuffle Read Size 的 max、75 分位和中位数；再看 Spill 和 GC Time。不要只看总时长。

解决方案：

加盐：给大表的热点 key 加随机前缀，把小表按前缀复制 N 份，join 后再聚合，热点 key 就被拆成 N 份。只对热点 key 加盐更好；全表加盐会让数据膨胀 N 倍。

两阶段聚合：先按 key + 随机盐 局部聚合，再去掉盐做第二次聚合。适合 GROUP BY 的 count/sum；求平均要同时保留 sum 和 count，最后再相除。具体代码见练习 2、3 和 3.2。

拆分热点 key：把空值、测试账号、头部商家单独过滤出来处理，正常 key 走常规逻辑，最后 union。如果空值没有业务意义，直接过滤最省事；如果有意义，给空值加随机后缀再聚合。

AQE：开启自适应执行后，Spark 会根据运行时 shuffle 统计合并小分区、切换 join 策略，并在检测到大分区时拆分。它是兜底，不是替代数据建模；热点 key 太离谱时仍要加盐或拆分。

2.3 Broadcast Join：把小表复制到每台机器

普通 join 要把两边按 key 重分区，产生 shuffle。如果一边很小，Spark 可以把它收集到 Driver，做成哈希表，广播到每个 Executor；大表不用动，每个分区本地查哈希表即可。

小表 -> Driver 收集 -> 广播到每个 Executor
大表的每个分区 + 小表哈希表 -> 本地 join（大表不 shuffle）

适用条件：

• 小表能完整放进每个 Executor 的内存。
• 默认阈值 spark.sql.autoBroadcastJoinThreshold=10MB，超过就不自动广播。
• 小表 x Executor 数量的网络复制成本可接受；几千个 Executor 广播 1 GB 表，网络也会爆。
• 小表是维度表、过滤后的结果、几百 MB 以内的字典表最合适。

from pyspark.sql.functions import broadcast

joined = items.join(
    broadcast(products.select("product_id", "category")),
    "product_id", "left",
)
joined.explain(True)

物理计划里应出现 BroadcastHashJoin 和 BroadcastExchange。如果还是 SortMergeJoin，检查表大小、统计信息、hint 是否被配置禁用。

注意：不要为了“避免 shuffle”强行广播几 GB 的表。Driver 收集会 OOM，Executor 复制会挤爆内存。广播是拿网络和内存换 shuffle，不是免费。

2.4 Cache、Persist 与 Checkpoint

detail.cache()
detail.count()        # 第一个 Action 才真正缓存
# ... 多次使用 detail ...
detail.unpersist()

spark.sparkContext.setCheckpointDir("data/checkpoint")
detail.checkpoint()

注意：

• cache() 是惰性的，必须有一次 Action 才物化。
• DataFrame 的 cache() 默认 MEMORY_AND_DISK；RDD 的 cache() 默认 MEMORY_ONLY。
• 缓存太多会把内存挤满，导致其他任务 spill；只缓存“多次使用且计算昂贵”的中间结果，用完 unpersist()。
• checkpoint 会写可靠存储，成本高；它不保留原血缘，适合超长 lineage 或需要截断恢复链的场景。
• checkpoint 目录要可写；Windows 本地写文件可能再次遇到 winutils.exe 问题。

2.5 序列化、内存、并行度与资源参数

序列化：RDD 默认用 Java 序列化，体积大、速度慢；可以换 Kryo：

spark = (SparkSession.builder
         .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
         .getOrCreate())

DataFrame 内部用 Tungsten 二进制格式，通常不受 Java 序列化影响；但 Python UDF 仍是逐行跨进程调用，能不用就不用。优先用 pyspark.sql.functions 内置函数；必须用 Python 逻辑时，考虑 pandas UDF/Arrow。

内存模型：Executor 内存大致分为预留内存、用户内存、Spark 统一内存（storage + execution）和堆外开销。spark.memory.fraction 默认 0.6，spark.memory.storageFraction 默认 0.5；execution 可以驱逐 storage，所以缓存不一定一直留在内存。spark.executor.memoryOverhead 默认约 10%，最小 384 MB，Python worker、JVM 内部结构都从这里出。

并行度：

• RDD 用 spark.default.parallelism；SQL shuffle 用 spark.sql.shuffle.partitions。
• 输入分区通常由文件块大小决定，spark.sql.files.maxPartitionBytes 默认 128 MB。
• 经验目标：每个分区 128–256 MB，总 Task 数约为总核数的 2–3 倍。
• repartition(n) 会 shuffle，可增可减；coalesce(n) 通常只减不增，不保证均衡。

资源参数（概念示例，本地不执行）：

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 4g \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 20 \
  --conf spark.sql.shuffle.partitions=400 \
  --conf spark.sql.adaptive.enabled=true \
  jobs/daily_gmv.py

怎么估：10 台机器，每台 16 核、64 GB，留 1 核和 8 GB 给系统，剩 15 核、56 GB；每个 Executor 5 核、18 GB 可放 3 个，共 30 个 Executor、150 核，shuffle 分区约 300–450。这是起点，不是固定答案。

Executor 不是越大越好：内存过大易 GC 停顿，核数过多会争抢单 JVM。常见做法是每 Executor 4–5 核、16–32 GB，留出 overhead；Driver 只在 collect()、广播、toPandas() 时压力大，不要因此调到几百 GB。

2.6 小文件与 AQE

小文件为什么麻烦：

• NameNode 为每个文件、每个块维护元数据，文件太多会吃内存。
• 每个文件可能对应一个 Task，启动和调度开销远大于读 1 MB 数据；listing 和查询规划也会变慢。
• 压缩文件如果不可切分（如单个大 gzip），并行度还会受限。

目标：单个文件接近 128–256 MB。常用手段：

• 写出前 repartition(n) 或 coalesce(n)，把文件数控制在合理范围。
• 不要对大数据 coalesce(1)：所有数据压到一个 Task，容易 OOM。
• write.option("maxRecordsPerFile", 500000) 控制单文件记录数。
• Hive 表可以开启小文件合并参数；Spark 写出后也可以再跑合并任务。
• 分区列基数不要太高：按天分区可以，按用户 ID 分区通常会产生海量小目录。

AQE（Adaptive Query Execution）：Spark 3.2 起默认开启。它在运行时根据 shuffle 统计调整计划：

常用参数：spark.sql.adaptive.enabled=true、spark.sql.adaptive.advisoryPartitionSizeInBytes=64m、spark.sql.adaptive.skewJoin.enabled=true；倾斜阈值参数（skewedPartitionFactor、skewedPartitionThresholdInBytes）随版本变化，以官方文档为准。AQE 能减少“分区数拍脑袋”的问题，但不能修复错误口径、不能减少源表扫描，也不能保证所有倾斜都被拆开。

3. 动手示例

3.1 构造一份倾斜数据

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = SparkSession.builder.appName("skew_demo").master("local[*]").getOrCreate()
spark.sparkContext.setLogLevel("ERROR")
skew = (spark.range(0, 1_000_000)
        .withColumn("key",
                    F.when(F.col("id") < 900_000, F.lit("hot"))
                     .otherwise(F.concat(F.lit("k"), (F.col("id") % 100).cast("string")))))
skew.groupBy("key").count().orderBy(F.desc("count")).show(5)

hot 约 90 万行，其余 100 个 key 各约 1,000 行。直接 groupBy("key").count()，会有一个 Task 处理 90 万行，其他 Task 很快结束。

3.2 两阶段聚合

N = 8
salted = skew.withColumn("salt", (F.rand() * N).cast("int"))
partial = salted.groupBy("key", "salt").count()
result = partial.groupBy("key").agg(F.sum("count").alias("cnt"))
result.orderBy(F.desc("cnt")).show(5)

第一阶段把 hot 随机分到 8 个盐值，热点被拆成 8 份；第二阶段再把 8 份加起来。结果和直接聚合一致，但单 Task 的数据量下降。盐值不是越多越好：N 太大，会产生更多小任务和额外 shuffle。

3.3 用 Spark UI 定位

本地运行时打开 http://localhost:4040（被占用会自动递增到 4041、4042）。重点看：

• Stages：每个 Stage 的 Task 数、输入大小、Shuffle Read/Write、Spill、GC Time。
• Summary Metrics：max、75 分位、median 的差距。最大值远高于中位数，基本就是倾斜。
• SQL：看执行计划里的 Exchange、BroadcastExchange、SortMergeJoin。
• Executors：每个 Executor 的内存、磁盘、失败任务数。

调优前先记录一组基线数字，改一个参数再对比。不要一次改十个参数，否则不知道哪个起了作用。

4. 常见坑

5. 练习

练习 1：从指标判断倾斜

某 Stage 有 200 个 Task，Shuffle Read 的中位数是 20 MB，最大值是 4 GB，且 3 个 Task 跑了 40 分钟，其余 2 分钟。判断问题并给出至少两种处理方向。

练习 2：写两阶段聚合

对 df 按 key 求 sum(amount)，用 16 个盐值写两阶段聚合。

N = 16
salted = df.withColumn("salt", (F.rand() * N).cast("int"))
partial = salted.groupBy("key", "salt").agg(F.sum("amount").alias("part_sum"))
result = partial.groupBy("key").agg(F.sum("part_sum").alias("total"))

练习 3：给大表 join 小表加盐

大表 facts 的 product_id 有热点，小维表 dim 只有几千行。写出加盐 join 的思路和关键代码。

N = 8
facts_salted = facts.withColumn("salt", (F.rand() * N).cast("int"))
dim_salted = dim.withColumn("salt", F.explode(F.sequence(F.lit(0), F.lit(N - 1))))
joined = facts_salted.join(F.broadcast(dim_salted), ["product_id", "salt"], "left")

练习 4：选广播 join 还是普通 join

两张表：orders 1,500 万行，customers 3,000 行。按 customer_id join，应该选哪种？怎么验证？有什么风险？

练习 5：估算 Executor 和分区

10 台机器，每台 16 核、64 GB。每台留 1 核和 8 GB 给系统，计划每个 Executor 5 核、18 GB。能放多少个 Executor？总核数多少？shuffle 分区大致设多少？

练习 6：修小文件

一个按天分区的表，每天写出 2,000 个 1 MB 小文件。写出至少三种处理办法，并说明哪种不能用于大数据。

6. 面试延伸

面试题：shuffle 为什么慢？

是什么：shuffle 是把数据按 key 重新分发的过程，发生在 groupBy、join、distinct、repartition、orderBy 等宽依赖处。

解决什么问题：它让相同 key 的数据聚到一起，才能聚合和 join；代价是 Map 端写磁盘、Reduce 端走网络拉取，还要序列化、反序列化和排序。

我的实操边界：我在本地观察过 Stage 切分和 Exchange 节点，理解 shuffle 的读写流程；没有处理过 TB 级 shuffle 的磁盘和网络瓶颈，参数调优只到“会看指标、会改常用项”的程度。

面试题：数据倾斜怎么处理？

是什么：少量 key 的数据量远大于其他 key，导致少数 Task 成为长尾，拖慢整个 Stage，严重时 OOM。

解决什么问题：通过打散热点、避免 shuffle 或拆分热点，让每个 Task 的数据量接近。

我的实操边界：我能在本地构造倾斜数据，用两阶段聚合和加盐验证效果；了解广播 join 和 AQE 的原理。生产上遇到过大促热点 key 的场景我还没有独立处理过，面试时会说明“方案清楚，生产实战有限”。

常用方案至少说两种：

1. 加盐打散：热点 key 加随机前缀，小表复制 N 份，join 后再聚合。
2. 两阶段聚合：先按 key + 盐局部聚合，再按业务 key 合并。
3. 广播小表：一大一小 join 时直接避免 shuffle。
4. 拆分/过滤热点 key：空值、测试账号、头部商家单独处理。
5. AQE 倾斜 join：运行时拆分超大分区，作为兜底。

面试题：cache 和 checkpoint 有什么区别？

是什么：cache/persist 把计算结果留在内存或磁盘，保留血缘；checkpoint 把结果写到可靠存储并截断血缘。

解决什么问题：cache 解决“同一份数据被多次使用”的重复计算；checkpoint 解决“血缘太长、重算代价高、故障恢复慢”。

我的实操边界：我在本地用过 cache() + unpersist()，也看过 checkpoint 目录；没有在生产上做过大规模缓存的容量规划和 checkpoint 恢复演练。

7. 本章验收

• 能画出 shuffle 的 Map 端和 Reduce 端流程，说出至少 3 项开销
• 能从 Stage 的 max/median 指标判断数据倾斜
• 能写出加盐 join 和两阶段聚合的代码
• 能说出广播 join 的适用条件和两个风险
• 能区分 cache/persist 与 checkpoint，知道各自适用场景
• 能按机器核数和内存估算 Executor 数量，并给出分区数起点
• 能说出小文件的危害和至少 3 种治理办法
• 能说出 AQE 的至少 3 项能力和它的边界
• 完成 6 道练习，至少 4 道能独立写出答案

下一章

数据采集入门：数据怎么从业务库、日志和埋点稳定地进入数仓。