Apache Spark中的DataFrame平等

Question

假设df1和df2是Apache Spark中的两个DataFrames，使用两种不同的机制计算，例如Spark SQL和Scala/Java/Python API。

是否有一种惯用的方法来确定两个数据帧是否相等（相等，同构），其中等价性是由数据（每行的列名和列值）相同（除了行的顺序）来确定的

这个问题的动机是，通常有很多方法来计算一些大数据结果，每种方法都有自己的权衡。在探索这些权衡时，重要的是保持正确性，因此需要检查有意义的测试数据集的等效性/相等性。

Answer 1

我不知道惯用语，但我认为您可以获得一种健壮的方法来比较数据帧，如下所述。（我使用PySpark进行说明，但这种方法可以跨语言使用。）

a = spark.range(5)
b = spark.range(5)

a_prime = a.groupBy(sorted(a.columns)).count()
b_prime = b.groupBy(sorted(b.columns)).count()

assert a_prime.subtract(b_prime).count() == b_prime.subtract(a_prime).count() == 0

这种方法可以正确处理数据帧可能具有重复行、不同顺序的行和/或不同顺序的列的情况。

例如：

python prettyprint-override">a = spark.createDataFrame([('nick', 30), ('bob', 40)], ['name', 'age'])
b = spark.createDataFrame([(40, 'bob'), (30, 'nick')], ['age', 'name'])
c = spark.createDataFrame([('nick', 30), ('bob', 40), ('nick', 30)], ['name', 'age'])

a_prime = a.groupBy(sorted(a.columns)).count()
b_prime = b.groupBy(sorted(b.columns)).count()
c_prime = c.groupBy(sorted(c.columns)).count()

assert a_prime.subtract(b_prime).count() == b_prime.subtract(a_prime).count() == 0
assert a_prime.subtract(c_prime).count() != 0

这种方法非常昂贵，但考虑到需要执行完全差异，大部分费用是不可避免的。这应该可以扩展，因为它不需要在本地收集任何东西。如果您放松了比较应该考虑重复行的限制，那么您可以删除groupBy（），只需执行subtract（），这可能会显著加快速度。

Answer 2

Apache Spark测试套件中有一些标准方法，但是其中大部分涉及在本地收集数据，如果您想在大型数据帧上进行相等测试，那么这可能不是一个合适的解决方案。

首先检查模式，然后可以与df3相交，并验证df1和df2的计数

另一个选项是获取两个数据帧的底层RDD，映射到（行，1），执行reduceByKey来计算每行的数量，然后将两个结果RDD进行组合，然后进行常规聚合，如果任何迭代器不相等，则返回false。

Answer 3

Scala（PySpark见下文）

spark fast测试库有两种进行数据帧比较的方法（我是该库的创建者）：

assertSmallDataFrameEquality方法在驱动程序节点上收集数据帧并进行比较

def assertSmallDataFrameEquality(actualDF: DataFrame, expectedDF: DataFrame): Unit = {
  if (!actualDF.schema.equals(expectedDF.schema)) {
    throw new DataFrameSchemaMismatch(schemaMismatchMessage(actualDF, expectedDF))
  }
  if (!actualDF.collect().sameElements(expectedDF.collect())) {
    throw new DataFrameContentMismatch(contentMismatchMessage(actualDF, expectedDF))
  }
}

assertLargeDataFrameEquality方法比较分布在多台机器上的数据帧（代码基本上是从spark testing base复制的）

def assertLargeDataFrameEquality(actualDF: DataFrame, expectedDF: DataFrame): Unit = {
  if (!actualDF.schema.equals(expectedDF.schema)) {
    throw new DataFrameSchemaMismatch(schemaMismatchMessage(actualDF, expectedDF))
  }
  try {
    actualDF.rdd.cache
    expectedDF.rdd.cache

    val actualCount = actualDF.rdd.count
    val expectedCount = expectedDF.rdd.count
    if (actualCount != expectedCount) {
      throw new DataFrameContentMismatch(countMismatchMessage(actualCount, expectedCount))
    }

    val expectedIndexValue = zipWithIndex(actualDF.rdd)
    val resultIndexValue = zipWithIndex(expectedDF.rdd)

    val unequalRDD = expectedIndexValue
      .join(resultIndexValue)
      .filter {
        case (idx, (r1, r2)) =>
          !(r1.equals(r2) || RowComparer.areRowsEqual(r1, r2, 0.0))
      }

    val maxUnequalRowsToShow = 10
    assertEmpty(unequalRDD.take(maxUnequalRowsToShow))

  } finally {
    actualDF.rdd.unpersist()
    expectedDF.rdd.unpersist()
  }
}

assertSmallDataFrameEquality对于小数据帧比较来说更快，我发现它对于我的测试套件来说已经足够了。

派斯帕克

下面是一个简单的函数，如果数据帧相等，它将返回true：

def are_dfs_equal(df1, df2):
    if df1.schema != df2.schema:
        return False
    if df1.collect() != df2.collect():
        return False
    return True

您通常会在测试套件中执行DataFrame相等性比较，并且在比较失败时需要一条描述性错误消息（调试时，True/False返回值没有太大帮助）。

使用chispa库访问返回测试套件工作流的描述性错误消息的assert_df_equality方法。