PySpark中dropDuplicates(subset)后调用exceptAll报错的解决方法

问题原因 #

当使用dropDuplicates(subset=[列名])时，Spark底层通过groupBy subset列 + 取每组第一行的逻辑实现去重。这个过程中，非subset列（如示例中的raw）会被重新计算生成新的内部列ID（如raw#1007），但原始列的ID（如raw#931）仍会被保留在执行计划中。而exceptAll操作依赖于DataFrame列的内部ID匹配（而非仅列名），导致Spark找不到原始列ID，抛出Couldn't find raw#xxx错误。

若不指定subset参数，dropDuplicates基于全列去重，不会修改列的内部ID，因此不会触发该错误。

解决方案 #

以下三种方法均可解决该问题，确保去重后的DataFrame能正常执行exceptAll：

1. 显式选择列重置元数据 #

在dropDuplicates后显式选择所有需要的列，强制Spark重新生成列的元数据，对齐两个DataFrame的列内部ID：

df1 = df1.dropDuplicates(subset=['n', 'm']).select('n', 'm', 'raw')
df1.exceptAll(df2).show()

2. 缓存去重后的DataFrame #

缓存操作会物化去重后的结果，截断之前的执行计划链，让后续操作使用新的列元数据：

df1 = df1.dropDuplicates(subset=['n', 'm']).cache()
df1.exceptAll(df2).show()

3. 手动实现groupBy+first去重 #

直接用groupBy搭配first函数实现subset去重，明确控制非subset列的取值，避免内部列ID混乱：

from pyspark.sql import functions as F
df1 = df1.groupBy('n', 'm').agg(F.first('raw').alias('raw'))
df1.exceptAll(df2).show()

验证结果 #

以上方案执行后，exceptAll都会返回预期的空DataFrame：

+---+---+---+
|  n|  m|raw|
+---+---+---+
+---+---+---+

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