如何逐列对比两个Spark DataFrame并存储不匹配记录？

原代码存在的问题 #

1. 语法错误：Scala关键字var/val需小写，toString首字母需大写
2. 逻辑缺陷：关联条件的字符串拼接无法正确识别字段不匹配关系
3. 性能隐患：多次调用count()和collect()会触发多轮Spark作业，且将大量数据拉至Driver端易引发内存溢出
4. 数据丢失：将DataFrame转为字符串存储再转回，会丢失原字段类型信息

优化后的实现方案 #

以下方案基于Spark DataFrame API实现，避免Driver端数据堆积，同时保留完整的字段类型信息：

1. 导入依赖包

import org.apache.spark.sql.functions._
import scala.collection.mutable.ArrayBuffer

2. 定义待对比的字段集合

排除关联键hash_key和row_num，只保留业务字段：

val columnsToCompare = srcTable_colMismatch.schema.fields
  .map(_.name)
  .filter(colName => !Set("hash_key", "row_num").contains(colName))

3. 关联两个DataFrame并标记不匹配字段

通过全外连接覆盖两边的新增/缺失记录，逐列对比并生成不匹配标记：

// 全外连接两个表，用hash_key和row_num作为关联依据
val joinedDF = srcTable_colMismatch.as("src")
  .join(
    tgtTable_colMismatch.as("tgt"),
    $"src.hash_key" === $"tgt.hash_key" && $"src.row_num" === $"tgt.row_num",
    "fullouter"
  )

// 为每个业务字段生成不匹配标记列（1表示不匹配，0表示匹配）
val mismatchFlagCols = columnsToCompare.map { colName =>
  when(
    $"src.$colName" =!= $"tgt.$colName" || 
    $"src.$colName".isNull =!= $"tgt.$colName".isNull,
    1
  ).otherwise(0).alias(s"${colName}_mismatch")
}

// 添加标记列，并生成全局不匹配标识
val withMismatchFlags = joinedDF
  .select(
    $"src.*", $"tgt.*", $"src.hash_key", $"src.row_num",
    mismatchFlagCols: _*
  )
  .withColumn("has_mismatch", greatest(mismatchFlagCols: _*))

4. 筛选并存储不匹配记录

方式一：直接保留DataFrame格式（推荐，大数据量友好）

// 筛选出存在任意字段不匹配的记录
val mismatchDF = withMismatchFlags.filter($"has_mismatch" === 1)

// 直接将DataFrame存储到变量，后续可直接用于分析或写入存储
mismatchDF.show()

方式二：收集到Driver端集合（仅适用于小数据集）

如果确实需要将不匹配记录的关键信息存入内存集合：

// 提取不匹配记录的关联键和不匹配字段名
val mismatchRecords = mismatchDF
  .select(
    $"hash_key",
    $"row_num",
    concat_ws(",", columnsToCompare.map(colName =>
      when(col(s"${colName}_mismatch") === 1, lit(colName)).otherwise(lit(null))
    ): _*).alias("mismatched_columns")
  )
  .collect()

// 存入ArrayBuffer
val mismatchBuffer = ArrayBuffer.from(mismatchRecords)

方案优势 #

• 仅触发一次Spark作业，性能远高于原代码的多轮作业模式
• 全外连接覆盖所有差异场景（字段值不匹配、单边新增记录）
• 保留原数据类型，避免字符串转换导致的信息丢失
• 仅在必要时才将数据拉至Driver端，降低内存溢出风险

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