PySpark Left Anti连接后空DataFrame却插入源数据的异常问题

核心错误定位 #

你的代码存在两个致命的Join语句错误，直接导致了异常现象：

1. 关联表写错：df1.join(df1, ...) 中右表错误使用了df1，而非需要对比的df2
2. 条件引用未关联的DataFrame列：Join条件里引用了df2的列，但df2并未参与本次Join操作，这会导致Spark执行计划逻辑混乱，甚至触发意外的执行偏差

异常现象解析 #

• df_Extract.show()显示空DataFrame：你用df1和自身做Left Anti Join，且条件中df2列属于非法引用，无法形成有效匹配，最终返回空结果
• 写入时插入df1全量数据：由于Join条件引用了未关联的df2列，Spark优化器无法正确解析逻辑，导致执行计划偏离预期，错误将原始df1数据写入目标表

修复后的代码 #

from pyspark.sql.functions import col

df1 = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:oracle:thin:@dsafsda:123/fd") \
    .option("query", """select * from employees""") \
    .load()

df2 = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:@das:132/wq") \
    .option("query", """select * from employees""") \
    .load()    

# 修正Join语句：右表改为df2，添加表别名避免列名冲突
df_Extract = df1.alias("a").join(
    df2.alias("b"),
    (col("a.emp_id") == col("b.emp_id")) & (col("a.dept_id") == col("b.dept_id")),
    "leftanti"
)

df_Extract.show()                                                          

df_Extract.write \
    .mode("append") \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:@das:132/wq") \
    .option("dbtable", "employee_new") \
    .save()    

job.commit()

额外排查建议 #

• 执行df_Extract.explain()查看逻辑计划和物理计划，确认Join操作是否符合预期
• 对不同来源的DataFrame添加表别名，避免列名冲突引发逻辑错误
• 写入前调用df_Extract.count()验证数据量，再执行写入操作，避免意外数据写入

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