Apache Spark中的DataFrame相等

Question

假设df1并且df2是DataFrameApache Spark 中的两个,使用两种不同的机制计算,例如,Spark SQL与Scala/Java/Python API.

是否存在一种惯用的方法来确定两个数据帧是否相等(相等,同构),其中等价由数据确定(每行的列名和列值)是否相同,除了行和列的排序？

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

Answer 1

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

首先检查模式,然后你可以与df3进行交集并验证df1,df2和df3的计数是否相等(但是这只有在没有重复行的情况下才有效,如果有不同的重复行,这个方法仍然可以返回true).

另一个选择是获取两个DataFrame的底层RDD,映射到(Row,1),执行reduceByKey来计算每个Row的数量,然后将两个生成的RDD组合在一起,然后进行常规聚合并返回false任何迭代器都不相等.

Answer 2

我不知道惯用语,但我认为你可以得到一种强大的方法来比较DataFrames,如下所述.(我使用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

此方法可以正确处理DataFrame可能具有重复行,不同顺序的行和/或不同顺序列的情况.

例如:

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 3

该火花快速测试库中有使数据帧比较（我是图书馆的创建者）两种方法：

该assertSmallDataFrameEquality方法在驱动程序节点上收集DataFrames并进行比较

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方法比较分布在多台计算机上的DataFrames（代码基本上是从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 对于较小的DataFrame比较，速度更快，并且我发现它足以用于我的测试套件。

Answer 4

爪哇：

assert resultDs.union(answerDs).distinct().count() == resultDs.intersect(answerDs).count();

Answer 5

尝试执行以下操作：

df1.except(df2).isEmpty

Answer 6

一种可扩展且简单的方法是比较两个DataFrames 并计算不匹配的行数：

df1.diff(df2).where($"diff" != "N").count

如果该数字不为零，则两个DataFrames 不相等。

该转换由spark-extensiondiff提供。

它标识已插入、已更改、已删除和已更改的行。

Answer 7

有 4 个选项，具体取决于您是否有重复的行。

假设我们有两个DataFrames，z1 和 z1。选项 1/2 适用于没有重复的行。您可以在 中尝试这些spark-shell。

• 选项1：直接执行 except

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.Column

def isEqual(left: DataFrame, right: DataFrame): Boolean = {
   if(left.columns.length != right.columns.length) return false // column lengths don't match
   if(left.count != right.count) return false // record count don't match
   return left.except(right).isEmpty && right.except(left).isEmpty
}

• 选项2：按列生成行哈希

def createHashColumn(df: DataFrame) : Column = {
   val colArr = df.columns
   md5(concat_ws("", (colArr.map(col(_))) : _*))
}

val z1SigDF = z1.select(col("index"), createHashColumn(z1).as("signature_z1"))
val z2SigDF = z2.select(col("index"), createHashColumn(z2).as("signature_z2"))
val joinDF = z1SigDF.join(z2SigDF, z1SigDF("index") === z2SigDF("index")).where($"signature_z1" =!= $"signature_z2").cache
// should be 0
joinDF.count

• 选项 3：使用GroupBy（对于具有重复行的 DataFrame）

val z1Grouped = z1.groupBy(z1.columns.map(c => z1(c)).toSeq : _*).count().withColumnRenamed("count", "recordRepeatCount")
val z2Grouped = z2.groupBy(z2.columns.map(c => z2(c)).toSeq : _*).count().withColumnRenamed("count", "recordRepeatCount")

val inZ1NotInZ2 = z1Grouped.except(z2Grouped).toDF()
val inZ2NotInZ1 = z2Grouped.except(z1Grouped).toDF()
// both should be size 0
inZ1NotInZ2.show
inZ2NotInZ1.show

• 选项 4，使用exceptAll，它也适用于具有重复行的数据

// Source Code: https://github.com/apache/spark/blob/50538600ec/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala#L2029
val inZ1NotInZ2 = z1.exceptAll(z2).toDF()
val inZ2NotInZ1 = z2.exceptAll(z1).toDF()
// same here, // both should be size 0
inZ1NotInZ2.show
inZ2NotInZ1.show