如何优化PySpark中exceptAll处理两个2GB数据文件的运行速度

1. 数据预处理优化 #

• 提前做列裁剪：如果仅需要按部分列判断重复，不要带全量列参与计算，仅保留判断重复的必要列和最终需要输出的列，大幅降低shuffle阶段的数据传输量

# 示例：仅用id、name两列判断重复，df2不需要保留额外输出列
df1 = df1.select("id", "name", "other_output_col1", "other_output_col2")
df2 = df2.select("id", "name")

• 提前过滤无效数据：先过滤掉空值、不符合业务规则的无用数据，减少参与计算的总数据量

df1 = df1.filter(col("id").isNotNull())
df2 = df2.filter(col("id").isNotNull())

2. 并行度与分区优化 #

• 调整shuffle分区数：默认spark.sql.shuffle.partitions值为200，你的总数据量仅4G左右，过多分区会产生大量小任务调度开销，建议调整为集群可用CPU核心数的2_{3倍，一般10}50区间即可

# 任务启动前配置
spark.conf.set("spark.sql.shuffle.partitions", 20)

• 按相同规则提前分区：对两个DataFrame按判断重复的列做相同规则的哈希/范围分区，避免执行exceptAll时再次触发全量shuffle

df1 = df1.repartition(20, "id", "name")
df2 = df2.repartition(20, "id", "name")

3. 执行策略替换优化 #

exceptAll底层需要对两个DataFrame做全量排序比对，shuffle开销更高，可替换为左反join实现等价逻辑，性能提升明显：

• 无需保留df1重复行的场景（即结果自动去重）：

df3 = df1.join(df2, on=["id", "name"], how="left_anti")

• 需要完全等价exceptAll逻辑（保留df1的重复行，比如df1某行出现3次、df2出现1次，最终返回2次）：

from pyspark.sql.window import Window
from pyspark.sql.functions import row_number

# 给df1相同行打上行号标记
w1 = Window.partitionBy("id", "name").orderBy("id")
df1_with_rn = df1.withColumn("rn", row_number().over(w1))

# 给df2相同行打上行号标记
w2 = Window.partitionBy("id", "name").orderBy("id")
df2_with_rn = df2.withColumn("rn", row_number().over(w2))

# 左反join匹配后删除辅助列
df3 = df1_with_rn.join(df2_with_rn, on=["id", "name", "rn"], how="left_anti").drop("rn")

4. 内存与广播优化 #

• 小表广播：你的df2仅2G，完全可以广播到所有Executor节点，彻底避免shuffle开销，这个优化收益最高

from pyspark.sql.functions import broadcast
# 左反join场景
df3 = df1.join(broadcast(df2), on=["id", "name"], how="left_anti")
# 坚持用exceptAll的场景也可以加广播
df3 = df1.exceptAll(broadcast(df2))

• 重复使用的DataFrame提前缓存：如果后续还要用到df1、df2，提前缓存到内存减少重复读文件开销

df1.cache()
df2.cache()
# 触发缓存
df1.count()
df2.count()

5. 其他细节优化 #

• 读文件时明确指定Schema，关闭自动Schema推断，减少文件读取阶段的耗时
• show()不要拉取全量数据，默认仅拉取20条即可，不要写df3.show(df3.count())这类全量拉取到Driver的逻辑
• 调整Standalone集群资源配置：给Executor分配不低于4G的内存，避免计算过程中数据溢写磁盘，尽可能打满集群可用CPU核心数。

内容的提问来源于stack exchange，提问作者TheBoredEcho