PySpark多轮连接操作性能优化求助：大表小表连接效率问题

核心问题拆解 #

• 小数据集（10条记录）与大表关联时，Spark未自动触发Broadcast Join，仍采用Sort Merge Join导致大表shuffle开销
• 后续小数据集关联时继承了大表的30个分区，过多分区引发不必要的任务调度与shuffle开销
• show()操作延迟是因为未缓存最终结果，重复执行计算逻辑

1. 强制广播小表，彻底消除大表shuffle #

显式标记小数据集为广播变量，强制Spark使用Broadcast Hash Join，大表无需shuffle，直接在Executor本地匹配数据：

from pyspark.sql.functions import broadcast

# 缓存小数据集，避免重复计算
mismatch_ids_row = (sourceonedf.join(sourcetwodf, on=primary_key, how='outer')
                    .where(condition)
                    .select(primary_key)
                    .cache())
mismatch_ids_row.count()  # 触发缓存加载

# 显式广播关联大表
df_1 = sourceonedf.join(broadcast(mismatch_ids_row), on=primary_key, how='inner').dropDuplicates()
df_2 = sourcetwodf.join(broadcast(mismatch_ids_row), on=primary_key, how='inner').dropDuplicates()

关键说明：手动广播绕过Spark自动判断逻辑，确保小表数据分发到所有Executor，大表本地完成匹配，IO开销骤降。

2. 压缩小数据集分区数，减少调度开销 #

两个小数据集（各10条左右）关联时，30个分区完全冗余，手动重分区为1个：

# 重分区小数据集，避免多任务空跑
df_1 = df_1.repartition(1)
df_2 = df_2.repartition(1)

# 执行full outer join
df = df_1.join(df_2, on=primary_key, how="full_outer")

关键说明：小数据量下单分区足够处理，消除多分区带来的任务调度、文件读写额外消耗。

3. 缓存最终结果，加速show()操作 #

通过缓存避免show()重复执行计算逻辑：

df.cache()
result_count = df.count()  # 触发全量计算并缓存
print(f"差异记录总数：{result_count}")
df.show()  # 直接读取缓存，秒级响应

关键说明：count()会触发Job执行并将数据写入内存缓存，后续show()无需重新计算。

4. 适配小数据场景的Spark配置优化 #

在任务启动时添加以下配置，进一步降低调度开销：

# 确保小表被自动识别为广播对象（可选，默认10MB足够）
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "10485760")
# 开启自适应分区合并，自动合并小任务
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionNum", "1")

关键说明：自适应分区合并会自动将数据量极小的分区合并，减少Executor任务调度次数。

预期效果 #

通过上述优化，可将整体耗时从500秒压缩至300秒以内，核心是避免大表shuffle、减少无效分区、消除重复计算三个方向。

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