在Databricks中如何正确对比两个Spark DataFrame的差异

问题1解答：对比前必须排序 #

因为Spark是分布式计算框架，DataFrame本身不保证行的存储/返回顺序，两份数据源即使数据完全一致，转成Pandas时的行顺序也可能不同，直接对比必然会出现误判。
排序操作可以选在Spark阶段或者Pandas阶段执行：

• Spark阶段排序（更推荐，大数据量下性能更好）：

# 按联合主键排序后再转Pandas
table_df = spark.sql("select * from db.table1").orderBy("ID", "Account")
file_df = spark.read.format("csv").load("my_file.txt", header = False, delimiter = '|').orderBy("ID", "Account")

• Pandas阶段排序：

# 排序后要重置索引，避免索引错位导致匹配错误
file_df = file_df.sort_values(by=["ID", "Account"]).reset_index(drop=True)
table_df = table_df.sort_values(by=["ID", "Account"]).reset_index(drop=True)

额外注意：你读取CSV时设置了header=False，此时CSV默认列名为_c0、_c1这类自动生成的名称，需要先手动指定和表数据完全一致的列名、对齐数据类型，否则后续对比也会出错。

问题2解答：行匹配逻辑 #

你当前用的Pandas eq()方法是按行索引位置逐行匹配，完全不识别业务主键：第一行表数据只会和第一行文件数据对比，第二行和第二行对比，和ID、Account字段没有任何关联。如果两边行顺序不一致，哪怕所有主键对应的数据完全一致，也会得到不一致的结果。
如果要基于联合主键匹配行，必须先通过主键做join关联，再对比非主键列。

问题3解答：当前方案不是最优方案 #

当前方案有三个明显缺陷：

• 全量数据转Pandas需要把所有数据拉取到Driver节点，数据量稍大就会触发Driver内存溢出，无法适配大数据场景
• 依赖行顺序匹配，容错率极低，很容易出现误判
• 只能返回全量布尔对比结果，无法快速定位具体是哪条主键的数据出现差异

更推荐直接在Spark层完成对比，不需要转Pandas：

# 第一步：先对齐CSV的列名、数据类型，确保和table_df的结构完全一致
file_df = file_df.toDF("ID", "Account", "col3", "col4") # 替换成实际的表列名
# 按需做数据类型转换，比如将ID列转成INT类型
file_df = file_df.withColumn("ID", file_df["ID"].cast("int"))

# 第二步：取双向差集得到所有差异数据
# 表存在、文件不存在的差异
diff_table = table_df.exceptAll(file_df)
# 文件存在、表不存在的差异
diff_file = file_df.exceptAll(table_df)
# 合并所有差异
all_diff = diff_table.unionByName(diff_file)

# 差异数为0则两份数据完全一致
print(f"差异数据条数：{all_diff.count()}")
# 可以直接查看具体差异数据
all_diff.show()

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