如何计算两个DataFrame各列的对应行差异计数？

要高效统计结构一致的百万级DataFrame每列对应行的数值差异数量，可通过关联DataFrame+逐列差异判断+聚合求和实现，具体步骤如下：

核心思路 #

1. 以唯一标识列（如示例中的City）或行号关联两个DataFrame，确保对应行匹配；
2. 对每一列判断两个DataFrame的对应值是否不同，将差异标记为1、相同标记为0；
3. 对每列的标记值求和，得到该列的总差异数。

代码实现（PySpark） #

假设已加载好df1和df2，且两表通过City列一一对应：

步骤1：重命名DF2的列，避免关联后列名冲突 #

from pyspark.sql.functions import col, when, sum

# 给DF2的所有列加上后缀，区分DF1的列
df2_renamed = df2.select([col(c).alias(f"{c}_df2") for c in df2.columns])

步骤2：按唯一标识关联两个DF #

# 以City为关联键，确保对应行匹配
joined_df = df1.join(df2_renamed, df1.City == df2_renamed.City_df2, how="inner")
# 移除重复的City列
joined_df = joined_df.drop("City_df2")

步骤3：逐列统计差异数量 #

# 遍历所有列，计算每列的差异计数
diff_counts = joined_df.select(
    *[sum(when(col(c) != col(f"{c}_df2"), 1).otherwise(0)).alias(c) for c in df1.columns]
)

# 查看结果
diff_counts.show()

无唯一标识时的处理方案 #

如果没有业务唯一键，可通过添加全局行号实现对应行关联：

from pyspark.sql.functions import monotonically_increasing_id

# 给两个DF添加行号
df1_with_id = df1.withColumn("row_id", monotonically_increasing_id())
df2_with_id = df2.withColumn("row_id", monotonically_increasing_id())

# 按行号关联
joined_df = df1_with_id.join(df2_with_id, on="row_id", how="inner").drop("row_id")

# 后续统计逻辑同步骤3

性能优化建议（针对百万级数据） #

• 分区优化：关联前对两个DF按关联键（如City）重分区，减少shuffle开销：

  df1 = df1.repartition("City")
  df2_renamed = df2_renamed.repartition("City_df2")
  

• 类型匹配：确保两DF对应列的数据类型完全一致，避免隐式类型转换影响性能；
• 过滤前置：如果存在无需比较的列，提前过滤减少计算量。

执行上述代码后，即可得到与示例一致的输出结果，且该方案基于Spark分布式计算，可高效支撑百万级数据处理。

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