PySpark中高效移除小DataFrame重复主键记录及广播连接疑问

首先明确：广播连接完全适用于你的场景。

为什么适用？ #

你的场景是小表（df_A，几MB）和大表（df_B，GB级）的关联过滤，这正是广播连接的典型适用场景——小表数据量极小，广播到所有执行节点的成本可以忽略，却能避免对大表进行昂贵的shuffle操作。

广播连接的核心作用 #

1. 彻底消除大表的shuffle开销
   原SQL中的NOT IN子查询，Spark默认可能会把df_B全量扫描后进行shuffle，将相同P值的数据分发到同一节点再和df_A做比对。而广播连接会把df_A的P值集合直接广播到每个执行节点，每个节点只需要用本地的df_B分片数据和广播过来的P集合做过滤，完全不需要移动df_B的海量数据，节省大量网络传输时间。

2. 大幅提升执行效率
   几MB的小表广播到所有节点几乎瞬间完成，每个节点可以独立处理本地的df_B数据片段，无需跨节点协同，对于GB级的df_B来说，这种本地化处理能把执行时间压缩到原方案的几分之一甚至更低。

3. 避免执行计划偏差
   虽然Spark Catalyst优化器通常会自动识别小表并启用广播，但手动指定广播提示（比如BROADCAST）可以确保优化器选择最优路径，不会因为表的统计信息不全或更新不及时导致错误选择低效的shuffle连接。

优化后的实现方案 #

更推荐用左反连接替代NOT IN（避免null值导致的异常结果），结合广播提示：

Spark SQL方式

SELECT *
FROM df_A_table a
WHERE NOT EXISTS (
    SELECT 1 FROM df_B_table b WHERE a.P = b.P
)
/*+ BROADCAST(a) */

DataFrame API方式（更直观）

from pyspark.sql.functions import broadcast

# 先对df_B的P去重，进一步减少待匹配的数据量
df_B_unique_p = df_B.select("P").distinct()

# 左反连接：保留df_A中不在df_B_unique_p里的记录，同时广播小表df_A
result_df = broadcast(df_A).join(df_B_unique_p, on="P", how="left_anti")

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