PySpark如何仅用一次Join同时实现Left Semi和Left Anti操作拆分大表

解决方案 #

可以通过一次Join实现需求，相比两次Join的方案能减少一半左右的大表扫描/Shuffle开销，更适合你当前的大数据量场景。

实现逻辑 #

你只需要给小表t1先加一个存在标记列，再用t2左关联处理后的t1，最后基于标记列拆分即可：

from pyspark.sql.functions import lit, broadcast

# 处理t1：先对id去重避免join后数据膨胀，添加匹配标记列
t1_with_flag = t1.select("id").distinct().withColumn("exists_flag", lit(True))
# 一次左关联，t1仅100万条属于小表，建议开广播完全避免Shuffle
joined = t2.join(broadcast(t1_with_flag), on="id", how="left")

# 拆分两个结果集，删除额外的标记列恢复原始t2的字段结构
existsDF = joined.filter(col("exists_flag") == True).drop("exists_flag")
absentDF = joined.filter(col("exists_flag").isNull()).drop("exists_flag")

方案对比 #

• 原两次Join方案：不管是否开启广播Join，都需要完整扫描两次10亿条的t2，如果没开广播还会产生两次t2的全量Shuffle，资源开销是一次Join方案的2倍。
• 单次Join方案：仅需扫描一次t2，开广播的情况下甚至不会产生Shuffle，整体执行效率明显更高。

额外优化建议 #

如果你的id存在数据倾斜问题，可以给t1的id加随机前缀打散后再关联，进一步避免少数task负载过高的问题。

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