企业官网建设流程全解析

前两篇把环境搭建、DataFrame 基础算子和最常见的数据清洗写法收拾完了，这篇我想直接聊点坐上生产环境后才会遇见的问题。很多人问我：“同一个任务，为什么我同事写得跑十几分钟，我写的要跑两个多小时？”除了资源差异，更常见的原因是读入阶段、Join 策略、UDF 序列化这几个位置，每个看起来都只是慢了一点，叠起来就是数量级的差距。

这篇“PySpark 秘籍（三）”里的坑，大多是从某套模拟项目的数据处理链路里复盘出来的，也参考了社区里频繁被讨论的典型误区。适合已经写过 PySpark、但总感觉差点意思的读者，尤其是准备做性能调优和问题排障的人。我不会把官方文档复述一遍，只讲我在实际排查里验证过的结论。

1. 数据读入阶段的决策，比后端的各种调参更值得先算清楚

很多优化工作其实在 action 之前就已经注定结果。第一次spark.read用什么格式、怎么切分文件、过滤条件有没有推给底层，直接决定了后续所有任务的上限。你是先读进来再慢慢处理，还是让 Spark 只读该读的部分，这两者的差别在数据量上来之后会非常恐怖。

1.1 分区数和文件块大小，决定了第一个“并行度”假象

先纠正一个常见的直觉：并行度不是越大越好。读 CSV 时 Spark 基本按文件分片，读 Parquet 时按文件或 row group 切分。如果你面对几百个几 KB 的小文件，启动 Stage 时会冒出大量碎任务，每个任务都在抢调度、序列化、建连接，真正干活的时间可能只有几十毫秒。反过来，一个 2GB 的压缩文件被切成太少的分片，又会造成某些 executor 空转，某些 executor 撑着。

实操上我一般把数据控制在单文件 128MB 到 512MB 左右。太小的文件提前合并；太大的文件用coalesce或重分区重新落盘。对于 Parquet，还可以按日期或者业务键做目录分区，让 Spark 只扫需要的目录。别小看这一步，有一次模拟项目里，仅仅是把几百个小 JSON 文件转成按天分区的 Parquet，同一个过滤查询从 40 秒降到了 6 秒。

还有一点很容易被忽视：CSV 和 JSON 的 schema 推断。inferSchema听起来省事，但它会让 Driver 提前扫描一部分数据来猜字段类型。字段一旦复杂，这个预扫描就可能占掉读取时间的相当比例。生产环境里我宁愿手写一个StructType，哪怕字段多写几分钟，换来的稳定性完全值得。嵌套 JSON 更是如此，字段结构稍微一改，推断结果就可能出现类型漂移。

1.2 谓词下推失效的三类现场

谓词下推的意思是，把where或filter里的条件尽可能压到数据源层，让底层只返回满足条件的文件、row group 或行。理解了这个机制，你就知道哪些写法会让它失效。

第一类常见现场：分区列被函数包住。比如F.year("event_date") == 2024，Spark 无法简单判断这个条件对应哪些分区目录，于是只能把全量分区读回来再过滤。正确写法是把时间范围写清楚：F.col("event_date") >= "2024-01-01" & F.col("event_date") < "2025-01-01"。前者能不能跑？能。但每次都要全扫。区别在千万行级别瞬间拉开。

第二类现场：过滤条件写在了 UDF 之后的临时列上。因为withColumn("new_col", my_udf(...))产生的列没有统计信息，Spark 通常不会把它推到数据源。更麻烦的是，如果你只是拿它做辅助判断，最好先算清楚“这个过滤真的需要在 Python 侧做吗”，很多情况用内置函数完全可以替代。

第三类现场：根本看不到下推效果。看完explain()的输出，如果PushedFilters里没有任何东西，不要猜，先检查你的数据源是不是视图、外部表或者拼接出来的临时表。某些连接器不会透传所有过滤条件，这时候与其在 Spark 侧硬顶，不如把过滤下沉到 SQL 层或者物化出一个中间结果。

1.3 读入阶段先看“三张图”

判断读入阶段有没有问题，我一般不看总运行时间，而是看 Spark UI 的 Stage 详情页。三个指标最值得盯：每任务Input大小是否均匀、Task Time是否有长尾、以及Shuffle Read是不是在读取阶段就反常出现。

如果某个 Stage 几乎全是Scan parquet和Scan json，但任务完成时间像锯齿一样参差不齐，多半是文件大小不均匀。如果任务时间都很短，但任务数量多到几千个，那是小文件问题。如果读个文件居然产生了几百 MB 的Shuffle Read，那就不是在读数据，是在读你前面某个算子广播出去的冗余结果。这时候先回头改读入，不要急着给 executor 加内存。

2. Join 慢的根子，多半不在“数据量大”，而在策略和键分布

Join 是 PySpark 里最容易被甩锅的场景。很多人一上来就说“数据量太大了，Spark 不行”，但点开执行计划一看，真正的问题根本不在数据量，而是 Spark 选了一个特别昂贵的关联策略，或者热点键把所有数据压进了同一个任务里。

2.1 先看懂 explain：BroadcastHashJoin 和 SortMergeJoin 的区别

任何 Join 排查的第一步都是df_a.join(df_b, ["key"]).explain("formatted")。输出里一般能看到两类策略：

• BroadcastHashJoin：把右侧数据集做成哈希表，广播到每个 executor 上，左表在原地完成关联。优点是没有 shuffle，速度快。
• SortMergeJoin：两侧数据先按 key 分别 shuffle、排序，再在每个分区里做归并。适合大表关联大表，但 shuffle 和排序成本都很高。

我见过不少小表关联反而慢的案例，其实就是右侧表序列化后的体积刚好超过广播阈值，Spark 自动选择了 SortMergeJoin。明明一张几 MB 的维度表，却要跟着主表一起 shuffle 两轮，慢几乎是一定的。

另一个细节是：广播阈值 10MB 是按“逻辑计划中该侧数据的预估大小”算的，不是按你脑子里想的“这张表在磁盘上多大”。如果右侧 DataFrame 经过多层withColumn、派生字段、字符串拼接，预估体积很容易膨胀。所以不要觉得表小就一定会广播。

2.2 手动广播和拆列，是成本最低的优化

确定要用广播，最简单就是broadcast()包一下：

from pyspark.sql.functions import broadcast result = orders.join(broadcast(dim_city), "city_id", "left")

但手动广播也有讲究。我一般先把维表裁剪到只留下 Join 键和要输出的几个字段，再做广播。少一列，广播体积就小一截；体积小，网络传输和 executor 内存占用都会下降。反过来，如果你把一个几百 GB 的表强制包进broadcast()，每个 executor 都要在内存里放一份，GC 压力比 SortMergeJoin 的 shuffle 还可怕。所以手动广播只适合“确实小”的表，别靠感觉。

2.3 数据倾斜现场：加盐二次聚合

数据倾斜太常见了。现象很典型：绝大多数任务秒级完成，唯独一两个任务卡在那里跑半小时。从 UI 上看，就是某个 Stage 的任务耗时图横着一根长条。

根因是热点 key。比如订单表的city_id90% 落在同一个城市，关联维度表时，这个 key 的所有数据都要进同一个 reducer。此时再多的 executor 都没用，因为单任务处理量已经超限。

常用的处理思路是加盐二次聚合。简单说，把热点 key 随机拆成 N 份，同时把维表按这 N 份展开，让原本压在一个任务里的数据分散到多个任务，最后再按真实 key 聚合回来。下面是一个简化的写法：

from pyspark.sql import functions as F # 假设 hot_keys 是统计出来的热点 key 列表 large_df = ... dim_df = ... hot_keys = ["city_001"] # 实际请通过 groupBy + 阈值筛选 hot_large = large_df.filter(F.col("city_id").isin(hot_keys)) hot_dim = dim_df.filter(F.col("city_id").isin(hot_keys)) # 对大表加随机盐，对小表按盐展开 salted_large = hot_large.withColumn("salt", (F.rand() * 10).cast("int")) salted_dim = hot_dim.withColumn( "salt", F.explode(F.array(*[F.lit(i) for i in range(10)])) ) hot_joined = ( salted_large.alias("l") .join(salted_dim.alias("d"), ["city_id", "salt"], "left") .groupBy("city_id") # 这里按业务聚合回结果 ) normal_joined = large_df.filter(~F.col("city_id").isin(hot_keys)).join( dim_df, "city_id", "left" ) result = normal_joined.unionByName(hot_joined)

注意，加盐方案最怕业务语义不对。随机盐会把一对一的明细关联变成多对多匹配，所以这个写法只适合关联之后再按 key 做聚合统计的场景。如果你关联完还要保留原始明细结构，别乱加盐，优先考虑让热点 key 单独走广播。

3. Window 函数慢起来，锅往往在“全局排序”和“过度排序”

Window 函数在 PySpark 里是“好用但贵”的典型。语法很简单，可是很多人写完之后发现任务时间暴涨。这时候最常见的错误是把责任推给窗口本身，其实问题出在排序和分区方式上。

3.1 一个常见写法：全局 row_number 当 row_id

F.row_number().over(Window.orderBy("ts"))这类写法，看起来是要给每行编一个全局递增序号。问题在于Window.orderBy("ts")没有写partitionBy，这意味着 Spark 需要把所有数据拿去做一次全局排序。数据量稍大，这个排序就会成为全任务的瓶颈。

如果只是需要一个唯一标识，完全没有必要排序，直接用F.monotonically_increasing_id()就好。它只是给行分配一个单调递增的 64 位整数，不保证连续，也不做全局排序，代价低得多。

如果你确实需要按某个字段排序后取序号，那应该先问自己：这个序号是不是只在某些分组内有意义？绝大多数业务里的“前几条”“最近一条”，其实都是分组需求，应该写成Window.partitionBy("user_id").orderBy("ts")，而不是对全表排序。少一个全局排序，往往就能把 Stage 时间从分钟级拉回秒级。

3.2 改成 groupBy + join，但要先检验业务语义

窗口函数最常被用于“取每个分组里时间最大的一条记录”。很多人顺手就写：

df.withColumn( "rn", F.row_number().over(Window.partitionBy("user_id").orderBy(F.col("ts").desc())) ).filter(F.col("rn") == 1).drop("rn")

能跑通，但如果这张表是几千万行，partitionBy和orderBy各自都会制造不小的 shuffle 和排序成本。有一种等价改写思路：先用groupBy拿到每个用户的最大时间，再把它作为 Join 条件回到原表：

max_ts = df.groupBy("user_id").agg(F.max("ts").alias("ts")) result = df.join(max_ts, ["user_id", "ts"]) \ .dropDuplicates(["user_id", "ts"])

这个写法把两轮 shuffle 变成一轮聚合加一轮关联，很多场景下会明显更快。但它有个前提：同一用户在同一时间点最多只保留一条记录，或者你要用dropDuplicates自己去重。如果窗口里还有row_number的“并列都保留”或“只保留第一条”的不同语义，改写前必须先用小数据验证再上线。

3.3 partitionBy 字段顺序和物理分布

Window.partitionBy里写多个字段时，字段顺序会直接影响哈希分区后的数据分布。比如partitionBy("device_type", "user_id")和partitionBy("user_id", "device_type")，虽然分区键组合相同，但 Spark 对分区键做哈希时，前序字段基数大、后续字段就容易被“淹没”。实际操作中，我一般会把基数高的用户维度放前面，把枚举型的字符串放后面，这样分布更均匀。

另外，窗口大量出现在同一个 Job 里时，要留意spark.sql.shuffle.partitions的默认值。200 个分区对于几亿行数据未必够，但也不是越大越好。分区数一旦超过实际可用的并发度，任务调度开销会抵消并行收益。建议根据数据量和 executor 核数动态估算，而不是一直用默认值到底。

4. Python UDF 一多，性能账单就失控：序列化、Arrow 和批量化的选择

PySpark 最微妙的地方在于跨语言边界。用户习惯写 Python，但 Spark 的核心执行引擎在 JVM 里。每次在 DataFrame 上调用一个普通 Python UDF，数据都要从 JVM 侧序列化出来，交给 Python 进程处理，再把结果序列化回去。一遍两遍可能感觉不到，一旦 UDF 出现在几千万行上，开销就会线性放大。

4.1 UDF 的隐形成本，比你想的贵得多

普通F.udf是逐行调用的，这是一个很容易被低估的事实。底层是一个“一行一对象”的转换循环，每行都要完成一次跨进程搬运。如果你在这个 UDF 里做了正则、JSON 解析、网络请求，那耗时就更是没法看了。

开启 Arrow 之后会好一些吗？要看场景。spark.sql.execution.arrow.pyspark.enabled=true主要改善的是toPandas()、pandas_udf这类批量列式传输场景，对普通 UDF 并不能带来明显提升。不少人在网上抄了这段配置，以为 UDF 就会自动变快，结果跑完发现进度条还是卡在原处，原因就在这里。

4.2 把 UDF 改写成列表达式，收益最直接

最常见的优化不是让 UDF 更快，而是让这个 UDF 不存在。

比如根据状态码生成描述，直接这样写：

from pyspark.sql import functions as F new_df = df.withColumn( "status_desc", F.when(F.col("status") == "1", "active") .when(F.col("status") == "2", "paused") .otherwise("unknown") )

字符串处理也一样，能用F.split、F.regexp_extract、F.trim、F.lower解决的，就不要进 Python。日期计算优先用F.date_add、F.datediff。这些内置表达式在 JVM 内直接运行，不需要序列化，也没有进程切换。

真正复杂的文本解析，可以考虑pandas_udf。它接收的不是单行，而是一批 pandas Series，批量化处理能摊薄跨语言开销。但注意returnType一定要写准确，否则容易踩到类型推断的坑。比如：

import pandas as pd from pyspark.sql.functions import pandas_udf @pandas_udf("string") def clean_text(s: pd.Series) -> pd.Series: return s.str.replace(r"\s+", " ", regex=True)

这种写法适合“逻辑确实复杂、改写成内置表达式很痛苦”的场景。我自己常用的判断标准是：如果函数在单行 pandas 上能轻松跑清楚，那就用pandas_udf；如果可以拆成when/otherwise组合，那就别碰 Python。

4.3 模型和 HTTP 资源的重复初始化，是另一种灾难

还有一类 UDF 慢不是函数体本身慢，而是每次调用都在重新加载资源。最常见的是在 UDF 里直接 load 一个机器学习模型，或者发一次 HTTP 请求。每行请求一次、每行加载一次模型，几万行下来就是几万次连接建立，Stage 不卡才怪。

我的做法是把资源初始化的次数从“每行”降为“每分区”。用mapInPandas是一个不错的选择，它允许你在一个分区数据上批量处理，并且只在新分区开始时做一次初始化。大概形态是：

from pyspark.sql import functions as F def process_partition(pdf_iter): model = load_model() # 每个分区只加载一次 for pdf in pdf_iter: pdf["pred"] = model.predict(pdf[["feature"]]) # 批量预测 yield pdf result = df.mapInPandas(process_partition, schema=df.schema.add("pred", "double"))

如果是小型模型并且只读不改，也可以用broadcast把模型对象广播出去，避免在每个 executor 上重复加载。但模型特别大的时候广播内存也扛不住，还是分区初始化更稳。

5. AQE 确实有用，但别把它当成“开了就快”

AQE 是 Spark 3.x 之后一个非常优秀的自适应执行能力。它可以在运行期根据已经完成的 Stage 统计信息，动态调整后续的执行计划。但我看到太多人把 AQE 当成了万能开关，配置一开就以为万事大吉，实际上它救不了所有问题。

5.1 先看 Spark UI 三个数据点

排查性能问题，我不看“总耗时”，我更习惯打开 Spark UI 的 SQL 页和 Stages 页，依次看三个数据点。

第一是Shuffle Read。如果某个 Stage 的 Shuffle Read 明显超大，说明上游产生了不必要的数据扩散，往往和 Join 策略、分区字段选择有关系。

第二是Shuffle Spill (disk)。这个指标一旦出现，说明内存已经不够装下 shuffle 数据了，数据被溢写到磁盘，之后还要读回来。这不是 AQE 能解决的，而是内存配置或分区粒度的问题。

第三是任务耗时直方图的长尾。整体耗时看着不高，但个别任务跑了几倍于中位数的时间，多半指向数据倾斜。AQE 的 skew join 能处理一部分，但前提是它被触发，且场景匹配。

5.2 AQE 优化了什么、没优化什么

AQE 主要做三件事：动态合并小分区、动态切换 Join 策略、动态处理 Join 倾斜。以常见的配置为例：

spark.conf.set("spark.sql.adaptive.enabled", "true") spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true") spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true") spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128M")

动态合并分区会把 shuffle 后过多的小任务合并成更合理的数量；动态切换 Join 策略会在运行期发现一侧数据足够小时改用 broadcast；skew join 则会在某些 key 严重倾斜时动态拆分。这些能力都非常有用。

但边界也要心里有数：AQE 只能影响“shuffle 之后”的执行计划，改不了读取阶段的小文件问题，也改不了 Python UDF 的序列化开销。如果瓶颈发生在读数据或者 UDF 上，开再多的 AQE 也只是让后半个任务快一点，前半个依旧原地踏步。

还有一个小坑：在某些强倾斜场景里，AQE 合并分区可能把分区数合并得特别少，反而导致并发度不够。遇到这种情况，可以给coalescePartitions.minPartitionNum设一个下限，让 Spark 不要无脑往少数分区里塞数据。

5.3 给资源估算留出余地，别只堆机器

最后说点生产上的体会。调优的顺序比调优本身更重要。先确认数据量级和 key 分布，再看执行计划，最后才轮得到配参数。直接堆 executor、加内存，常常只是把一段慢任务变成了另一段比较贵但没那么慢的任务。

我一般会在任务跑之前根据数据量估一下 shuffle 分区数，把 AQE 开着，但保留一个合理的advisoryPartitionSizeInBytes，再随手检查 Spark UI。把读入阶段、Join 策略、UDF 边界和 AQE 边界都过一遍之后，大多数“PySpark 慢”的问题其实都能落到某一个具体环节上，而不是玄学。道理想通了，参数只是最后的落地动作。