Apache Spark 中的 DataFrame 相等性

Question

假设 df1 和 df2 是 Apache Spark 中的两个 DataFrames，使用两种不同的机制计算得出，例如 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

Scala（PySpark 见下文）

spark-fast-tests 库有两种进行 DataFrame 比较的方法（我是该库的创建者）：

assertSmallDataFrameEquality方法收集驱动节点上的DataFrame并进行比较

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 比较更快，我发现它对于我的测试套件来说已经足够了。

PySpark

这是一个简单的函数，如果 DataFrame 相等则返回 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 方法，该方法返回测试套件工作流的描述性错误消息。

Answer 4

您可以结合使用一点重复数据删除和完全外连接来完成此操作。这种方法的优点是它不需要您将结果收集到驱动程序，并且可以避免运行多个作业。

import org.apache.spark.sql._
import org.apache.spark.sql.functions._

// Generate some random data.
def random(n: Int, s: Long) = {
  spark.range(n).select(
    (rand(s) * 10000).cast("int").as("a"),
    (rand(s + 5) * 1000).cast("int").as("b"))
}
val df1 = random(10000000, 34)
val df2 = random(10000000, 17)

// Move all the keys into a struct (to make handling nulls easy), deduplicate the given dataset
// and count the rows per key.
def dedup(df: Dataset[Row]): Dataset[Row] = {
  df.select(struct(df.columns.map(col): _*).as("key"))
    .groupBy($"key")
    .agg(count(lit(1)).as("row_count"))
}

// Deduplicate the inputs and join them using a full outer join. The result can contain
// the following things:
// 1. Both keys are not null (and thus equal), and the row counts are the same. The dataset
//    is the same for the given key.
// 2. Both keys are not null (and thus equal), and the row counts are not the same. The dataset
//    contains the same keys.
// 3. Only the right key is not null.
// 4. Only the left key is not null.
val joined = dedup(df1).as("l").join(dedup(df2).as("r"), $"l.key" === $"r.key", "full")

// Summarize the differences.
val summary = joined.select(
  count(when($"l.key".isNotNull && $"r.key".isNotNull && $"r.row_count" === $"l.row_count", 1)).as("left_right_same_rc"),
  count(when($"l.key".isNotNull && $"r.key".isNotNull && $"r.row_count" =!= $"l.row_count", 1)).as("left_right_different_rc"),
  count(when($"l.key".isNotNull && $"r.key".isNull, 1)).as("left_only"),
  count(when($"l.key".isNull && $"r.key".isNotNull, 1)).as("right_only"))
summary.show()

Answer 5

Java：

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

Answer 6

尝试执行以下操作：

df1.except(df2).isEmpty

Answer 7

try {
  return ds1.union(ds2)
          .groupBy(columns(ds1, ds1.columns()))
          .count()
          .filter("count % 2 > 0")
          .count()
      == 0;
} catch (Exception e) {
  return false;
}

Column[] columns(Dataset<Row> ds, String... columnNames) {
List<Column> l = new ArrayList<>();
for (String cn : columnNames) {
  l.add(ds.col(cn));
}
return l.stream().toArray(Column[]::new);}

columns 方法是补充，可以替换为任何返回 Seq 的方法

逻辑：

1. 合并两个数据集，如果列不匹配，则会抛出异常并因此返回 false。
2. 如果列匹配，则对所有列进行 groupBy 并添加列计数。现在，所有行的计数都是 2 的倍数（即使是重复行）。
3. 检查是否有任何行的计数不能被 2 整除，这些是多余的行。

Answer 8

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

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

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

diff 转换由spark-extension 提供。

它识别I插入、C挂起、D删除和uN更改的行。 p>

Answer 9

根据您是否有重复行，有 4 个选项。

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

• 选项 1：直接执行除外

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