如何在PySpark中高效判断两个大型DataFrame是否相等？

针对100GB级别的大型DataFrame，绝对不能用toPandas()把全量数据拉到Driver内存（这就是你内存溢出的原因），必须用分布式计算的方式生成校验和或快速校验，以下是几种可行方案：

方案1：使用Spark内置校验和函数（Spark 3.1+推荐） #

Spark 3.1及以上版本提供了checksum()函数，可以直接对整个DataFrame计算全局校验和，完全分布式执行，不会占用Driver过多内存：

from pyspark.sql.functions import checksum

# 计算df1的全局校验和
df1_checksum = df1.select(checksum(*df1.columns)).first()[0]
# 计算df2的全局校验和
df2_checksum = df2.select(checksum(*df2.columns)).first()[0]

# 对比校验和
if df1_checksum == df2_checksum:
    print("两个DataFrame校验和一致，大概率无差异")
else:
    print("两个DataFrame校验和不一致，存在差异")

注意：checksum()会考虑所有列的内容和顺序，列顺序不同也会得到不同结果，如果需要忽略列顺序，可以先统一列顺序再计算。

方案2：分布式行哈希聚合 #

如果你的Spark版本低于3.1，可以手动对每行生成哈希，再通过分布式聚合得到全局哈希值：

from pyspark.sql.functions import sha2, concat_ws, col

# 把所有列拼接成字符串，生成每行的SHA256哈希
row_hashes_df1 = df1.select(sha2(concat_ws("|", *df1.columns), 256).alias("row_hash"))
row_hashes_df2 = df2.select(sha2(concat_ws("|", *df2.columns), 256).alias("row_hash"))

# 对所有哈希值做全局聚合（排序后拼接再哈希，避免顺序影响结果）
def aggregate_hashes(hashes):
    import hashlib
    combined = "".join(sorted(hashes))
    return hashlib.sha256(combined.encode()).hexdigest()

# 先按分区聚合哈希，再全局聚合，减少Driver内存压力
def partition_hash(iterator):
    import hashlib
    combined = "".join(sorted(iterator))
    return [hashlib.sha256(combined.encode()).hexdigest()]

# 计算df1全局哈希
df1_part_hashes = row_hashes_df1.rdd.map(lambda x: x[0]).mapPartitions(partition_hash).collect()
df1_global_hash = aggregate_hashes(df1_part_hashes)

# 计算df2全局哈希
df2_part_hashes = row_hashes_df2.rdd.map(lambda x: x[0]).mapPartitions(partition_hash).collect()
df2_global_hash = aggregate_hashes(df2_part_hashes)

if df1_global_hash == df2_global_hash:
    print("全局哈希一致，数据大概率无差异")
else:
    print("全局哈希不一致，数据存在差异")

方案3：快速元数据+统计信息校验（前置过滤） #

在计算校验和之前，可以先做快速校验，快速排除明显的差异：

• 元数据校验：对比两个DataFrame的列名、列类型、行数是否完全一致：

  # 对比列名和类型
  if set(df1.dtypes) != set(df2.dtypes):
      print("列名或列类型不一致，存在差异")
  # 对比行数
  elif df1.count() != df2.count():
      print("行数不一致，存在差异")
  else:
      print("元数据一致，继续校验内容")
  

• 统计信息校验：对数值列对比统计量（count、mean、min、max），对分类列对比distinct数量：

  # 获取统计信息并对比
  df1_stats = df1.describe().collect()
  df2_stats = df2.describe().collect()
  
  if df1_stats != df2_stats:
      print("统计信息不一致，存在差异")
  

这种方法速度极快，但存在小概率的假阴性（统计量相同但数据不同），适合作为初步过滤，再结合校验和方法做最终确认。

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