Spark & PySpark
几乎每一个慢速 Spark 作业都逃不过三个原因之一：一次本可避免的 shuffle、一个倾斜的键、或者一行行通过 Python 慢速处理的数据。最大的错误是在阅读执行计划之前就调整集群规模——花钱买更多的 executor 来跑同一个低效的 DAG。
操作流程
步骤 1：收集输入
- 作业代码，对于慢速作业，还需要 Spark UI 中慢速阶段的视图（任务持续时间分布、shuffle 读写大小）以及
df.explain(True)的输出。 - 输入数据的大小和格式，以及每个连接两侧数据的大小。
- 集群配置：executor 数量、核心数、内存。
- 作业是新建还是调优：对于调优，先通过执行计划诊断，再修改代码。
步骤 2：从合理的会话开始
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
spark = (
SparkSession.builder
.appName("etl")
.config("spark.sql.shuffle.partitions", "200")
.config("spark.sql.adaptive.enabled", "true")
.getOrCreate()
)
启用自适应查询执行（AQE）；它会在运行时合并 shuffle 分区并切换连接策略。然后仔细设置 spark.sql.shuffle.partitions：目标为每个 shuffle 分区 100-200 MB，因此分区数 ≈ shuffle 数据大小 ÷ 128 MB，并且至少是总 executor 核心数的 2-3 倍，以确保没有核心空闲。
步骤 3：坚持使用 DataFrame API，远离 Python
DataFrame 操作会经过 Catalyst 和 Tungsten，从而获得查询优化和堆外内存。仅当没有 DataFrame 原语可用时才降级到 RDD。Python UDF 会将每一行序列化到 Python 工作进程再返回，破坏了 codegen。优先级顺序：
- 使用内置的
pyspark.sql.functions（例如F.regexp_extract,F.when）。 - 当无法避免 Python 时，使用 Pandas（向量化）UDF。
- 仅作为最后手段使用标量 Python UDF。
from pyspark.sql.functions import pandas_udf
@pandas_udf("double")
def normalize(s):
return (s - s.mean()) / s.std()
步骤 4：根据大小选择连接策略
- 当较小的表可以放入内存时，对其进行广播。Spark 会自动广播低于
spark.sql.autoBroadcastJoinThreshold（默认 10 MB）的表；当 executor 内存允许时，可以显式广播大小达到几百 MB 的表，但大约 1 GB 是实际天花板——广播会将表复制到每个 executor，如果判断失误会导致作业 OOM。
from pyspark.sql.functions import broadcast
result = large.join(broadcast(small), "key")
- 在连接之前过滤列和行，以减少 shuffle 数据量；每减少一列和一行就节省了 shuffle 的字节数。
- 两个大表 → 接受 sort-merge 连接，但先检查是否存在数据倾斜（步骤 5）。
步骤 5：诊断并修复数据倾斜
数据倾斜的症状：阶段中只剩下一两个任务未完成；最大任务持续时间超过中位数的约 3-5 倍；Spark UI 中某个 shuffle 读取分区远大于其他分区。AQE 的倾斜连接处理会在分区大小超过中位数大小的 5 倍且超过 256 MB（默认值）时拆分该分区——在执行计划中确认它是否已触发。当 AQE 无法帮助时（例如倾斜聚合），对热点键进行加盐处理：对大表一侧的键追加随机后缀 0..N，将小表一侧在所有 N 个后缀上展开，连接，然后去除盐。选择 N ≈ 倾斜因子（如果某个键是中位数的 20 倍，则 N=20）。
步骤 6：控制分区和输出布局
repartition(n, col)会进行全 shuffle 以平衡数据；在宽输出之前使用。coalesce(n)减少分区但不进行 shuffle；用于避免产生过小的输出文件。- 目标输出文件大小为 128 MB - 1 GB；成千上万的小文件会拖慢每个下游读取器。
- 仅按低基数列（日期、区域——几十到几百个值，绝不使用 ID）进行分区输出：
df.write.partitionBy("dt").mode("overwrite").parquet("/data/out")
步骤 7：仅缓存被重用的数据
仅当 DataFrame 在 DAG 中被多次使用时才缓存，将其物化，并在完成后取消缓存：
df.cache()
df.count() # 物化
# ... 重用 df ...
df.unpersist()
缓存一次使用的 DataFrame 会浪费内存，并可能驱逐真正重要的数据。
步骤 8：通过执行计划验证
重新运行 df.explain(True)，确认预期的策略已经出现（BroadcastHashJoin、下推的过滤器、没有出现本应被消除的 Exchange）。读取列式格式（Parquet、ORC）以便谓词下推能够工作，永远不要对大量数据使用 collect() 将其拉取到 driver，并且在链式调用中优先使用 F.col 引用而不是字符串列名。
示例：糟糕的连接 vs 好的连接
糟糕的——一个 2 TB 的事件表连接一个 200 MB 的维度表，全列宽度，使用 Python UDF 进行解析：
result = (events.join(dims, "store_id") # sort-merge：shuffle 了全部 2 TB
.withColumn("region", parse_region_udf("meta")) # 逐行的 Python 处理
.filter(F.col("dt") == "2024-06-01")) # 连接之后才过滤
执行计划显示两侧都有 Exchange（shuffle 了 2 TB），并且 UDF 阻止了 codegen。
好的——先过滤和投影，广播维度表，使用内置函数：
result = (events
.filter(F.col("dt") == "2024-06-01") # 首先裁剪到约 30 GB
.select("store_id", "amount", "meta")
.join(broadcast(dims.select("store_id", "region_code")), "store_id")
.withColumn("region", F.regexp_extract("meta", r"region=(\w+)", 1)))
现在执行计划显示 BroadcastHashJoin，没有大表的 Exchange：扫描约 30 GB，大表零 shuffle，没有 Python 边界。
交付物
生成修订后的作业代码以及一份调优说明，内容应包括：每个连接的策略及其大小依据、shuffle 分区设置及其计算过程、发现的任何数据倾斜（最大与中位数任务比值）及修复措施、缓存决策、输出文件大小预期，以及（对于调优）调优前后的 explain 差异。
禁止事项
- 不要广播一个你没有估算大小的表——一个错误的广播会瞬间导致所有 executor OOM。
- 在检查
pyspark.sql.functions之前不要编写 Python UDF；几乎所有字符串、日期和条件逻辑都有内置函数。 - 不要对一个大型 DataFrame 使用
collect()或toPandas()；它会把集群的所有数据都灌入一个 driver。 - 不要按高基数列对输出进行分区——会产生数百万个目录，每个目录只有一行。
- 不要因为一个倾斜的任务导致慢速阶段就增加 executor；更多的核心无法拆分一个分区。
- 不要出于习惯任意缓存；只缓存那些在后续步骤中被重用的 DataFrame，并在使用后取消缓存。
质量标准
一个完成的作业只有在满足以下条件时才算是合格的：每个连接都标明了其策略以及该策略的大小依据；spark.sql.shuffle.partitions 是根据数据大小计算得出的，而不是沿用习惯值；Spark UI 中 shuffle 阶段的任何任务持续时间不超过中位数持续时间的约 3 倍（或者已解释并修复了倾斜）；任何可用的内置函数替代了标量 Python UDF；输出文件大小落在 128 MB - 1 GB 范围内；并且最终的 explain 确认了预期的执行计划。