如何在PySpark中基于concern_code外连接多DataFrame并保留唯一主键列

现有代码问题 #

你的代码存在两个核心错误：

• 未显式指定连接类型，PySpark的join方法默认使用内连接(inner)，无法满足外连接需求
• 从df3开始的连接使用列等值判断作为连接条件，而非公共列名列表，会同时保留左右表的concern_code列导致重复；且第一次join后中间结果已经不存在df1这个原始表别名，直接调用df1["concern_code"]会触发字段不存在的运行错误

最优修复方案 #

统一使用公共列名列表作为连接条件，同时显式指定外连接类型即可。PySpark识别到相同名称的公共连接键时，会自动合并为一列，最终仅保留一个concern_code：

# 多表全外连接标准写法
result_df = df1.join(df2, on=["concern_code"], how="outer") \
               .join(df3, on=["concern_code"], how="outer") \
               .join(df4, on=["concern_code"], how="outer") \
               .join(df5, on=["concern_code"], how="outer")

# 输出结果
result_df.show()

特殊场景兼容写法 #

如果有业务需求必须用等值条件写连接逻辑，需要先给原始df1设置全局别名，每次join后主动删除对应表的重复主键列：

from pyspark.sql.functions import col

result_df = df1.alias("base_df") \
               .join(df2, on=["concern_code"], how="outer") \
               .join(df3, col("base_df.concern_code") == df3["concern_code"], how="outer") \
               .drop(df3["concern_code"]) \
               .join(df4, col("base_df.concern_code") == df4["concern_code"], how="outer") \
               .drop(df4["concern_code"]) \
               .join(df5, col("base_df.concern_code") == df5["concern_code"], how="outer") \
               .drop(df5["concern_code"])

result_df.show()

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