Spark中如何验证两个DataFrame数据及行数一致性（ScalaTest场景）

问题背景 #

使用Spark 3、Scala 2.12.12及ScalaTest做测试，要验证get_amounts方法生成的resDf和预期df_expected的数据（包括行数）完全一致。Schema匹配测试已通过，但基于exceptAll的自定义assertData函数验证数据一致性时测试失败。

相关代码 #

定义Schema #

// 定义各DataFrame的Schema
val amountsSchema: StructType = StructType(
  Seq(
    StructField("positionId", LongType, nullable = true),
    StructField("amount", DecimalType(10, 2), nullable = true),
    StructField("eventTime",LongType, nullable = true),
  )
)

val warehouseSchema: StructType = StructType(
  Seq(
    StructField("positionId", LongType, nullable = true),
    StructField("warehouse", StringType, nullable = true),
    StructField("product", StringType, nullable = true),
    StructField("eventTime",LongType, nullable = true),
  )
)

val expectedDfSchema: StructType = StructType(
  Seq(
    StructField("positionId", LongType, nullable = true),
    StructField("warehouse", StringType, nullable = true),
    StructField("product", StringType, nullable = true),
    StructField("amount", DecimalType(10, 2), nullable = true),
  )
)

测试数据与DataFrame创建 #

// 测试数据
val amounts_data = Seq(
  Row("1", "5.00", "1528463387"),
  Row("1", "7.20", "1528463005"),
  Row("2", "5.00", "1528463097"),
  Row("2", "7.20", "1528463007"),
  Row("3", "6.00", "1528463078"),
  Row("4", "24.20", "1528463008"),
  Row("4", "15.00", "1528463100"),
)

val wh_data = Seq(
  Row("1", "W-1", "P-1", "1528463098"),
  Row("2", "W-2", "P-2", "1528463097"),
  Row("3", "W-2", "P-3", "1528463078"),
  Row("4", "W-1", "P-6", "1528463100"),
)

val expected_data = Seq(
  Row("1", "W-1", "P-1", "5.00"),
  Row("2", "W-2", "P-2", "5.00"),
  Row("3", "W-2", "P-3", "6.00"),
  Row("4", "W-1", "P-6", "15.00")
)

// 创建DataFrame
val df_amounts: DataFrame = spark.createDataFrame(
  spark.sparkContext.parallelize(amounts_data),
  amountsSchema
)

val df_wh: DataFrame = spark.createDataFrame(
  spark.sparkContext.parallelize(wh_data),
  warehouseSchema
)

val df_expected: DataFrame = spark.createDataFrame(
  spark.sparkContext.parallelize(expected_data),
  expectedDfSchema
)

// 生成结果DataFrame
val resDf: DataFrame = get_amounts(df_amounts, df_wh)

测试代码 #

// Schema匹配测试（已通过）
test("DataFrame Schema Test") {
  assert(assertSchema(resDf.schema, df_expected.schema))
}

// 数据一致性测试（失败）
test("DataFrame Data Test") {
  assert(assertData(resDf, df_expected))
}

// 自定义数据断言函数
def assertData(df1: DataFrame, df2: DataFrame): Boolean = {
  df1.exceptAll(df2).rdd.isEmpty()
}

排查与解决步骤 #

1. 修复测试数据类型不匹配问题 #

创建测试数据时用了字符串值，但Schema定义的是LongType和DecimalType，Spark自动转换可能导致隐性类型差异（比如Decimal的精度存储问题）。修改测试数据为对应原生类型：

val amounts_data = Seq(
  Row(1L, BigDecimal("5.00"), 1528463387L),
  Row(1L, BigDecimal("7.20"), 1528463005L),
  Row(2L, BigDecimal("5.00"), 1528463097L),
  Row(2L, BigDecimal("7.20"), 1528463007L),
  Row(3L, BigDecimal("6.00"), 1528463078L),
  Row(4L, BigDecimal("24.20"), 1528463008L),
  Row(4L, BigDecimal("15.00"), 1528463100L),
)

val wh_data = Seq(
  Row(1L, "W-1", "P-1", 1528463098L),
  Row(2L, "W-2", "P-2", 1528463097L),
  Row(3L, "W-2", "P-3", 1528463078L),
  Row(4L, "W-1", "P-6", 1528463100L),
)

val expected_data = Seq(
  Row(1L, "W-1", "P-1", BigDecimal("5.00")),
  Row(2L, "W-2", "P-2", BigDecimal("5.00")),
  Row(3L, "W-2", "P-3", BigDecimal("6.00")),
  Row(4L, "W-1", "P-6", BigDecimal("15.00"))
)

2. 对齐DataFrame字段顺序 #

exceptAll会严格匹配字段顺序，若resDf的字段顺序与df_expected不一致，会导致匹配失败。在比较前对齐字段顺序：

def assertData(df1: DataFrame, df2: DataFrame): Boolean = {
  // 对齐字段顺序
  val df1Aligned = df1.select(df2.columns.head, df2.columns.tail: _*)
  val df2Aligned = df2.select(df2.columns.head, df2.columns.tail: _*)
  // 双向检查差异
  val diffInDf1 = df1Aligned.exceptAll(df2Aligned)
  val diffInDf2 = df2Aligned.exceptAll(df1Aligned)
  diffInDf1.isEmpty && diffInDf2.isEmpty
}

3. 增加调试信息定位差异 #

修改assertData函数打印差异行，直观看到不匹配的数据：

def assertData(df1: DataFrame, df2: DataFrame): Boolean = {
  val df1Aligned = df1.select(df2.columns.head, df2.columns.tail: _*)
  val df2Aligned = df2.select(df2.columns.head, df2.columns.tail: _*)
  
  val diff1 = df1Aligned.exceptAll(df2Aligned)
  val diff2 = df2Aligned.exceptAll(df1Aligned)
  
  if (!diff1.isEmpty) {
    println("结果DataFrame独有的行：")
    diff1.show(false)
  }
  if (!diff2.isEmpty) {
    println("预期DataFrame独有的行：")
    diff2.show(false)
  }
  
  diff1.isEmpty && diff2.isEmpty
}

4. 验证get_amounts业务逻辑 #

从测试数据的预期结果来看，逻辑应该是为每个positionId匹配warehouse数据，并筛选出与warehouse的eventTime最匹配的amount值（比如晚于或等于该时间的最近记录）。检查get_amounts中的关联、窗口函数或筛选条件是否正确实现了这一逻辑。

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