企业官网建设流程全解析

1. 开篇：PySpark 真正的性能瓶颈，往往不在你的代码

前两篇把 DataFrame 的常用 API 和编码习惯讲得差不多了，这一篇我换个角度，聊跑离线批任务时最让人崩溃的场景：任务跑到一半 executor 挂掉、某个 stage 卡住不动、明明加了核心和内存还是不加速，以及频繁出现的 FetchFailedException。这些问题有一个共同特点：报错往往不直接指向真正的原因，需要一层层往下摸。

我的习惯是，接手一个 PySpark 作业时，先不看代码逻辑，先看两个东西：当前任务的执行计划里有多少 Shuffle，以及资源配置和分区数是否匹配。为什么要先看这两样？因为在大多数生产环境里，慢和挂的根源不是某个表达式写错，而是 Shuffle 放大了数据规模，或者资源参数与集群状态不匹配。这一篇的内容都围绕一个目标——把 Shuffle 管好、把分区和内存对齐、把排障路径缩短。适合的人群是正在写 PySpark 批处理任务、被性能和稳定问题反复折磨的工程师。里面的例子尽量贴真实场景，多数是云厂商集群上跑批处理的通用经验。

1.1 先打开 AQE：性价比最高的三个开关

Spark 3.0 之后，自适应查询执行（Adaptive Query Execution，也就是 AQE）是我在任何新建作业里都会开的功能。它不像很多人想象的是一个不稳定特性，在 3.2 之后的版本里它默认就是开启的。如果你还停留在 Spark 2.x 上，别想着靠写法和参数硬扛，换 3.3+ 比折腾其他东西划算得多。

AQE 的核心开关有三个：

• spark.sql.adaptive.enabled：总开关，不开这个后面两个都无效。
• spark.sql.adaptive.coalescePartitions.enabled：让 Spark 在每次 shuffle 之后根据真实输出大小自动合并过小的分区。
• spark.sql.adaptive.skewJoin.enabled：让 Spark 在运行时自动检测 join 一侧的数据倾斜，并把倾斜分区拆分。

把它们打开之后，Spark 会在每个 stage 执行完之后根据真实数据大小重新规划下一步的分区数，也会在 join 的时候自动处理比较明显的倾斜 key。对 PySpark 用户来说，这个功能特别有价值，因为 PySpark 里写出的 DataFrame 操作往往依赖 Spark 的优化器来修正；你很难用手工 repartition 去覆盖所有数据分布的波动。

from pyspark.sql import SparkSession spark = SparkSession.builder \ .appName("py_spark_cheatsheet_3") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ .config("spark.sql.adaptive.skewJoin.enabled", "true") \ .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64MB") \ .config("spark.sql.adaptive.coalescePartitions.parallelismFirst", "true") \ .getOrCreate()

这几个参数怎么定？advisoryPartitionSizeInBytes 是 AQE 合并小分区时的目标分区大小，我习惯先按每个 executor 的可执行内存除以并发度估算。比如单 executor 给 8G 内存、4 个 core，每个 task 大概能用 1.5G 左右，那我通常把目标分区大小设在 64MB 到 128MB 之间。这里的关键是不要让分区数少到单 task 内存不够，也不要多到 Shuffle 数据被切成几万个小文件。设置太大会导致 task 跑得又慢又容易 OOM；设置太小会让调度开销增长。曾经有一次我把这个值改成 128MB，结果某个 join 的每个 task 平均处理 1.2G 数据，加上计算侧的临时对象，反复触发 GC。改回 64MB 之后就好了。

AQE 对我来说最大的价值在于“同一条 job 今天数据和明天数据分布不同也能自动适应”，不需要每次上线前根据采样重新调整 repartition。尤其是数据量不稳定、上游分区数经常变的场景，开 AQE 能省大量开发时间。

1.2 内存账：executor 内存、Python worker 内存和 overhead 不是一回事

PySpark 新手最容易搞混的一件事是：Spark executor 的 JVM 内存和 Python worker 进程可用的内存是两个池子。PySpark 里 DataFrame 的 Java 侧对象在 JVM 堆内存里管理，而 Python UDF 执行时，数据会通过 PyArrow 序列化传到 Python worker 进程里去计算。如果 Python 侧内存不够，你看到的报错会是 Python worker exited unexpectedly，而不是常见的 Java heap space OOM。

我维护的一个批处理作业是这样的：executor 配置为 4 cores、8G 内存。最初没有单独设置 spark.python.worker.memory，结果一张 2 亿行的表做 groupBy 后再做 Python 侧自定义逻辑时，worker 反复崩溃，Spark 自动重试三次后整个 job 失败。后来我把 executor 内存调整为 6G，spark.python.worker.memory 独立预留 2G，才算稳定下来。配置参数的逻辑是：

• spark.executor.memory：JVM 堆内存，主要给 DataFrame 的 Java 侧算子使用。
• spark.python.worker.memory：每个 Python worker 的可用内存，不是总内存，是每个 worker 进程。
• spark.executor.memoryOverhead：off-heap 的额外内存，用于本地缓冲、网络栈、以及部分序列化缓冲等。

在提交参数时，我常这样设：

spark-submit \ --master yarn \ --executor-memory 6g \ --conf spark.python.worker.memory=2g \ --conf spark.executor.memoryOverhead=1g \ --num-executors 20 \ --executor-cores 4 \ job.py

这不是唯一的正确配置，但三者的关系需要分开想清楚。一个常见的错误是以为调大 executor memory 就能解决所有内存问题，结果调大之后 GC 时间更长；真正的问题在 Python worker 的内存被 UDF 吃光了。反过来，如果 UDF 不多，Python worker 内存留 1G 足够，不需要多分。

还要注意 partition 数对内存的影响。DataFrame 默认的 shuffle 分区数是 200，但很多场景下这根本不是最优值。我通常用一个很朴素的公式来估算：每个 task 处理的数据量不要超过单个 executor 可用内存的四分之一。比如目标分区大小 64MB、executor 可用内存 6G，那么每个 task 处理 64MB 数据完全没问题；如果某张表是 1T 的数据，分区数至少应该在 4096 以上，否则单个 task 拉取几百兆数据，reducer 侧的内存就会吃紧。分区数不是越大越好，但比数据量少太多时，内存问题会比序列化开销更早出现。

2. 数据倾斜：90% 的 OOM 都出在这

我先说一个现象：某个作业前半段跑得飞快，某个 stage 开始以后，进度条卡在 99%，任务日志里某些 task 处理时间明显比其他 task 长一个数量级；再翻 executor 日志，OOM 正好出现在那个 stage。十次里有八九次是数据倾斜。倾斜的本质是少数 key 对应的数据量远大于平均值，所以无论你怎么加 executor，只要这几个 key 固定落到某几个 task 上，就一直是单点瓶颈。

以前我在一个虚构的某运营数据仓库取数任务里遇到过类似情况：按城市维度 join 一份维表，某几个头部城市的订单量占了六成以上，输出结果时那一个 reduce task 要处理其余 task 几百倍的数据，整个 job 从半小时变成六小时。我后来总结出一套相对稳定的处理路径。

2.1 快速定位倾斜 Key 的方法

不要上来就猜，先用五分钟把一个 stage 里最可疑的 key 拎出来。最直接的方法是在 Spark UI 的 Stage 页面看每个 task 的 Input Size 和 Records：如果某个 task 的输入是其他 task 的几十倍，基本就是倾斜。接着针对具体的 join 或 groupBy 操作做一次单独统计：

from pyspark.sql import functions as F skew_summary = (df .groupBy("city_id") .agg(F.count("*").alias("cnt")) .orderBy(F.col("cnt").desc()) .limit(20)) skew_summary.show(20)

这条代码执行的是 groupBy 聚合，如果 city_id 分布均匀，结果里的最大值和一个普通值之间差距不会特别夸张；如果 top1 占了总量一半以上，接下来处理这张表之前就要先想倾斜方案。很多同学觉得跑这条统计也需要时间，但比起在倾斜 stage 里反复重试，这一步几乎可以忽略不计。

定位之后不要急着改代码。先分清楚倾斜的 key 是自然业务上就重的，比如热门城市、大客户，还是因为数据本身重复产生的空值或默认值。如果是空值或默认值，比如 user_id 为 null 或者填了 0 的记录，简单粗暴的解法是把这类 key 的取值做随机打散；如果是实打实的高频业务 key，就要考虑下面的方案之一。

2.2 两阶段聚合实战：加盐写法与适用边界

先给出一个标准的 groupBy 两阶段聚合示例。核心思想是：第一层用随机盐值把大 key 拆散，让数据分散到不同 task 上做预聚合，第二层去掉盐值，做最终聚合。这样原本压在一个 task 上的压力被摊到 N 个 task 上。

from pyspark.sql import functions as F salt_num = 50 # 第一阶段：加盐后按 原key + 盐值 聚合 stage1 = (df .withColumn("salt", F.rand() * salt_num) .groupBy("city_id", "salt") .agg(F.sum("order_amount").alias("part_amount"))) # 第二阶段：去掉盐值，最终聚合 result = (stage1 .groupBy("city_id") .agg(F.sum("part_amount").alias("total_amount")))

这里有一个细节：加盐字段我用 F.rand() * salt_num 而不是直接 F.rand()，是希望盐值落在 0 到 salt_num 之间连续分布；配合 groupBy 时，Spark 会根据这个字段把同一个原 key 的记录尽量分散。salt_num 怎么选？我一般取 50 到 100 之间，太小打散效果不明显，太大反而让第一层聚合的 shuffle 数据成倍增加。关键判断标准是：拆分后单 task 的数据量降到平均数据量的 10 倍以内。

但两阶段聚合只适用于按原 key 做聚合的场景。如果你是因为 join 而倾斜，两阶段聚合不能直接用，只能用广播或 AQE 的 skewed join。此外，如果倾斜 key 的分布本身不均匀，比如 top1 和 top2 相差悬殊，两阶段聚合依然可能让 top1 key 的预聚合 task 压力偏大，这时可以把 salt_num 提高，或者对单独的大 key 做特殊处理，而不是对所有 key 统一加盐。我在处理真正的热门 city_id 时，会先把 top 1 的 key 单独拆出来加盐，其余 key 保持正常路径，最终用 union 合并回去。

2.3 什么时候可以直接靠 AQE 自动处理 join 倾斜

Spark 3.2 之后，打开 spark.sql.adaptive.skewJoin.enabled，AQE 会在运行时自动检测 join 另一侧有没有某个分区的数据量特别大，然后把这个倾斜分区拆成多个小分区去 join。这个机制处理的是“某个分区太大导致 join 端压力集中”的情况，具体参数是：

• spark.sql.adaptive.skewJoin.skewedPartitionFactor：单个分区数据量达到中位数的多少倍被认为是倾斜，默认 5 倍。
• spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes：触发倾斜的最小数据量，默认 256MB。

不过我的经验是 AQE 自动倾斜 join 也不是万能的。它的原理依赖运行时统计，如果上游 stage 的数据量不大，倾斜阈值没有达到，它不会触发；而 groupBy 场景下的倾斜它管不了。所以别再迷信“开了 AQE 就万事大吉”，倾斜检查仍然要在代码里做。AQE 适合的是 join 场景里数据量已经比较可观的倾斜，如果倾斜特别严重，仍然需要手动拆 key。

3. 别到处写 UDF：内置函数和 pandas UDF 才是提速正道

在 PySpark 项目里，我见过太多团队把 PySpark 当 pandas 用。碰到一个日期字符串想转成时间差，写一个 def parse_date(s): return ...，然后注册成 UDF；要计算某个字符串是否包含关键词，也写 UDF。不是说 UDF 不能用，而是很多用 UDF 的场景，根本就是内置函数一行就能搞定的事。UDF 一旦用多，作业运行时间会成倍增加。

3.1 Python UDF 为什么慢的核心原因

先理解一下为什么 Python UDF 慢。普通的 DataFrame 算子，比如 filter、groupBy、join，Spark 内部走的是 JVM 侧的优化流程，数据以列式格式保存在 Java 内存里，计算也尽量向量化。而一旦用了 UDF，每行数据都要从 JVM 侧经过序列化传给 Python worker，Python 侧执行函数，再把结果序列化回 JVM。这里每行都有一次跨进程通信和两次序列化开销，所以数据量越大，拖慢越明显。有人比喻为：把一堆货卸下车、搬到另一个仓库逐件处理，再搬回来。显然这比直接在原仓库里分拣慢得多。

所以我在写 PySpark 代码的第一原则是：凡是用 DataFrame API 能完成的，绝不用 UDF。字符串处理、时间处理、数值计算、条件分支，绝大多数都有内置函数。

3.2 内置函数改写场景：日期、枚举、窗口统计

举一个最常见的场景。业务表里有一个 ts 字段，格式是 "2025-05-01 12:33:44"，要过滤出最近三天的记录。新手写法可能是：

from pyspark.sql.functions import udf from datetime import datetime, timedelta @udf("boolean") def is_recent(ts): try: d = datetime.strptime(ts, "%Y-%m-%d %H:%M:%S") return d >= datetime.now() - timedelta(days=3) except: return False df.filter(is_recent(F.col("ts"))).show()

用内置函数可以完全改成：

from pyspark.sql import functions as F df = df.withColumn("ts_dt", F.to_timestamp("ts", "yyyy-MM-dd HH:mm:ss")) df = df.filter(F.col("ts_dt") >= F.date_sub(F.current_timestamp(), 3))

to_timestamp、unix_timestamp、date_add、datediff 这类函数直接把大部分时间解析需求覆盖了。字符串场景同理，contains、startswith、regexp_replace、substring_index 在 PySpark 里都有。需要根据条件生成枚举值时，用 when 是首选，它比 UDF 快很多且易读：

level_col = F.when(F.col("cnt") > 1000, "high") \ .when(F.col("cnt") > 100, "medium") \ .otherwise("low")

另一个高频场景是分组 TopN。很多人会想到先 collect 再排序，其实用窗口函数就能优雅实现：

from pyspark.sql import Window w = Window.partitionBy("city_id").orderBy(F.desc("amount")) df = df.withColumn("rk", F.row_number().over(w)) top3 = df.filter(F.col("rk") <= 3)

我还见过用 UDF 做中位数计算的，这也别做，percentile_approx 是现成的。

3.3 pandas UDF 的真实收益和适用边界

如果确实有一段逻辑内置函数拼不出来，必须用自定义函数逐行或逐组处理，也不要直接写普通 UDF。pandas UDF 的收益在于：它在 Python 侧接收的是 pandas Series 或 DataFrame 批量数据，而不是单行，PyArrow 序列化开销被大幅摊薄。比如要做一段比较复杂的字符串清洗逻辑，可以用 pandas UDF 写成：

from pyspark.sql.functions import pandas_udf import pandas as pd @pandas_udf("string") def clean_text(s: pd.Series) -> pd.Series: # 一些复杂的 pandas 向量化处理 return s.str.strip().str.replace(r"\s+", " ", regex=True)

这样也比普通 UDF 快不少，但仍然比内置函数慢一个量级。以前一个文本特征提取函数用普通 UDF 跑一张 5000 万行的表要接近两个小时，改成 pandas UDF 后大约 25 分钟。注意，这里有个前提：集群的 PySpark 环境必须装了 pandas 和 pyarrow，否则 pandas UDF 会退化成普通 UDF 或者直接报错。还有一个坑是 pandas UDF 的返回类型必须稳定匹配，比如返回 Decimal 或数组类型时要特别注意 schema 申明。如果逻辑是按组处理，可以用 groupBy().applyInPandas 或 mapInPandas，但我个人建议：能先在 Spark 侧完成 groupBy 和窗口计算的，尽量留在 Spark 侧，不要让 Python 承载太重。

4. 高频报错排查速查表：现场别慌

无论调优做得多好，生产环境里总会遇到各种报错。这一节我把这些年积累的五个高频异常整理成速查表，方便你现场直接对照。

4.1 高频异常的分辨与处置

表格中列的都是我真实遇到的高频项。其中 FetchFailedException 特别容易混淆：它表面上是拉取失败，但根因经常是写 shuffle 的 executor 因为 OOM 挂了，所以发现这类报错时不要一股脑去调网络重试，先检查前后的内存异常。

4.2 一次 “Python worker exited unexpectedly” 的排障链

我拿一次真实作业来演示排查顺序。某天凌晨跑一个报表预处理任务，数据量比平时多了一倍多，作业在某个 groupBy 后的 map 阶段反复失败，错误日志频繁出现 Python worker exited unexpectedly (crashed)。第一反应是检查 Python worker 内存，把 spark.python.worker.memory 从 1g 调到 2g，结果发现仍然失败。

接下来看 Spark UI 里失败的 task 输入记录数和大小，发现每个 task 的输入记录数接近 80 万行。这个数量并不算大，继续看 worker 日志，发现进程是被系统 OOM killer 杀掉的，而不是 Spark 自己抛错；说明这个节点上的物理内存已经不足以承载这么多并发 Python worker 了。调用栈里是自定义 UDF 内部创建了一个比较大的中间字典，每个 worker 都会加载同样的字典，内存占用被放大。

最后的解法是分两步：一是把 UDF 改成 pandas UDF，避免每行都做跨进程序列化；二是把自定义字典改用只在 executor 初始化一次的方式加载，而不是在每个 UDF 内部每次重建。这两步调整后任务从失败变成稳定运行。这个案例说明一个问题：排障不能只照着报错文本改内存，还要看 worker 日志、executor 的资源使用曲线，以及代码里是否存在重复且高成本的初始化。

5. 提交作业前必走的几个检查点

最后写几个管用的习惯，都是踩过坑之后养成的。

5.1 explain 看物理计划，而不是 show 猜

写完一个 DataFrame 之后，不要只用 show 验证结果对不对，至少跑一次 explain。PySpark 里直接：

df.explain(mode="formatted")

我会重点看两个点：一是 join 和 groupBy 前后是否有 Exchange，也就是 shuffle 节点；二是每个 Filter 是否真的被下推到数据源。如果某张表读取时没有尽早过滤，导致全量数据进入后续计算，再好的调优也是白搭。见过不少本来可以推给上游的几亿行数据，因为用字符串拼接条件或者在 DataFrame 上先 join 后再过滤，造成 join 双方数据量爆炸。

5.2 cache、collect、toPandas 的节制原则

再说 cache 的边界。cache 是为了防止同一个 DataFrame 在多个 action 中被重复计算；如果只在一个 action 里用，cache 完全没必要。每次调 cache 前要确认后续至少有两个 action 会复用，否则只是白白占用内存。checkpoint 我很少用，它不是 cache 的替代品，主要用于切断血缘；用之前想清楚为什么要断血缘。如果为了调试，把超大 DataFrame toPandas 到本地也可能把 driver 内存打爆；真想看样例数据，用 head(100) 已经是最大容忍度。

5.3 版本选择：Spark 3.3+ 的默认值已经够省心

最后一个建议：如果条件允许，直接使用 Spark 3.3 或 3.5 版本。现代版本里 AQE 默认开、动态分区裁剪默认开，很多手工 repartition 和广播变量都变得可有可无。PySpark 用户最大的困惑往往不是 API，而是旧集群上那些过时的默认行为。升级大版本之后，很多曾经手动调的东西反而没必要了。我现在的习惯是：把所有写在作业里的手动配置圈出来，逐个问一句“现在还需要吗”。能删就删，让引擎自己去决定，往往才是更稳的方案。

如果你正在给一个老作业调优，我的建议是别一次性改一大堆参数。先把内存账算清楚，再开 AQE，再处理数据倾斜，最后才碰 UDF。每改一步，跑一次小数据验证，确认有效再放开全量。这样即使出了新问题，你也知道是哪个改动引起的。PySpark 的调优没有银弹，但顺着这条链路走，大部分作业都能从“勉强能跑”变成“稳定又快”。