单元测试中如何有效比较Spark Dataframe以验证转换功能正确性

1. 直接用assertEquals对比DataFrame：Spark的DataFrame没有重写equals方法，默认对比的是对象引用而非数据内容，所以就算数据完全一致也会失败。
2. except对比逻辑写反：两个DataFrame内容一致的前提下，actualResult.except(expectedResult).count()和expectedResult.except(actualResult).count()返回值都是0，你的断言条件判断不等于0，自然会失败。同时except方法会自动对结果去重，如果你的数据存在重复行，会出现误判。

方案1：原生Spark实现无依赖对比 #

不需要引入第三方库，自己实现对比逻辑，适合轻量测试场景：

• 先校验Schema是否一致，避免字段顺序、类型不匹配的隐性问题
• 用exceptAll替代except，不会去重，适配有重复行的业务场景
• 需要忽略行顺序的话，可先按全字段排序后再对比

@Test
def testPivotingDataFramePositive(): Unit = {
    val inputDf = spark.read.option("header",true).csv("src/test/resources/input-test-files/pivotInput.csv")
    val expectedResult = spark.read.option("header",true).csv("src/test/resources/output-test-files/pivotOutput.csv")
    val actualResult = parserObj.pivotingDataFrame(inputDf)
    
    // 第一步：对比Schema
    assert(actualResult.schema.equals(expectedResult.schema), "输出Schema与预期不符")
    // 第二步：对比数据内容
    assert(actualResult.exceptAll(expectedResult).count() == 0, "实际结果存在预期外的记录")
    assert(expectedResult.exceptAll(actualResult).count() == 0, "预期结果存在实际缺失的记录")
    
    // 如果要求行顺序严格一致，加上排序逻辑后再对比
    // val sortCols = actualResult.columns.sorted.map(col)
    // val sortedActual = actualResult.sort(sortCols:_*)
    // val sortedExpected = expectedResult.sort(sortCols:_*)
    // 再用上述exceptAll逻辑对比排序后的结果即可
}

方案2：用成熟工具类简化测试（多测试用例场景更高效） #

如果后续要写大量DataFrame测试用例，推荐用现成的测试工具，不用自己维护对比逻辑：

• 方式1：用Spark自带的测试基类DatasetSuiteBase，内置checkAnswer、checkDatasetEquals方法，自动处理Schema校验、行顺序、精度兼容等问题

import org.apache.spark.sql.test.DatasetSuiteBase

class ParserTest extends DatasetSuiteBase {
  @Test
  def testPivotingDataFramePositive(): Unit = {
    val inputDf = spark.read.option("header",true).csv("src/test/resources/input-test-files/pivotInput.csv")
    val expectedResult = spark.read.option("header",true).csv("src/test/resources/output-test-files/pivotOutput.csv")
    val actualResult = parserObj.pivotingDataFrame(inputDf)
    // 直接调用内置方法对比即可
    checkAnswer(actualResult, expectedResult.collectAsList())
  }
}

• 方式2：用spark-fast-tests第三方库，对比逻辑更灵活，支持自定义精度、忽略字段等能力

import com.github.mrpowers.spark.fast.tests.DataFrameComparer

class ParserTest extends DataFrameComparer {
  @Test
  def testPivotingDataFramePositive(): Unit = {
    val inputDf = spark.read.option("header",true).csv("src/test/resources/input-test-files/pivotInput.csv")
    val expectedResult = spark.read.option("header",true).csv("src/test/resources/output-test-files/pivotOutput.csv")
    val actualResult = parserObj.pivotingDataFrame(inputDf)
    // 小数据量用这个方法，大数据量可以用assertLargeDataFrameEquality性能更好
    assertSmallDataFrameEquality(actualResult, expectedResult)
  }
}

• 读取CSV文件时建议手动指定Schema，避免自动推断类型出错（比如数字被推断为字符串、空值默认类型异常等）导致不必要的测试失败
• 如果有浮点型字段，要配置精度容忍范围，避免计算精度误差导致测试不通过，上述两种工具类都支持自定义精度参数

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