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 内存调整为 6Gspark.python.worker.memory 独立预留 2G才算稳定下来。配置参数的逻辑是spark.executor.memoryJVM 堆内存主要给 DataFrame 的 Java 侧算子使用。spark.python.worker.memory每个 Python worker 的可用内存不是总内存是每个 worker 进程。spark.executor.memoryOverheadoff-heap 的额外内存用于本地缓冲、网络栈、以及部分序列化缓冲等。在提交参数时我常这样设spark-submit \ --master yarn \ --executor-memory 6g \ --conf spark.python.worker.memory2g \ --conf spark.executor.memoryOverhead1g \ --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.enabledAQE 会在运行时自动检测 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、joinSpark 内部走的是 JVM 侧的优化流程数据以列式格式保存在 Java 内存里计算也尽量向量化。而一旦用了 UDF每行数据都要从 JVM 侧经过序列化传给 Python workerPython 侧执行函数再把结果序列化回 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(days3) 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, , regexTrue)这样也比普通 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 高频异常的分辨与处置报错信息常见原因处理动作java.net.ConnectException: Connection refusedexecutor 进程异常退出或端口不可达多和内存有关看 worker 日志确认 OOM调大 spark.executor.memoryOverhead 或 python.worker.memory减少每个 executor 的 coreorg.apache.spark.shuffle.FetchFailedExceptionshuffle 数据拉取失败常常由上游 executor 丢失或磁盘损坏导致先看是不是网络或节点故障稳定后重试如果频繁出现则降低分区数SparkOutOfMemoryError: Java heap spaceJVM 侧内存不够常见于大 shuffle 或广播变量过大检查分区数是否过少改用 AQE减小广播数据集调大 executor memoryPython worker exited unexpectedlyPython worker 进程被杀死多为 worker 内存不足调大 spark.python.worker.memory减少 UDF 侧单任务数据量Cannot pickle自定义对象无法序列化常见于 closure 中传入不可 pickle 对象避免把 SparkSession、连接池等对象放进 UDF 闭包改成传递必要参数表格中列的都是我真实遇到的高频项。其中 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(modeformatted)我会重点看两个点一是 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 的调优没有银弹但顺着这条链路走大部分作业都能从“勉强能跑”变成“稳定又快”。
