Равномерность DataFrame в Apache Spark

Ответ 1

В наборах Apache Spark есть несколько стандартных способов, однако большинство из них предполагает сбор данных локально, и если вы хотите провести тестирование равенства на больших DataFrames, то это, вероятно, не подходит.

Сначала проверьте схему, а затем вы можете сделать перекресток с df3 и убедиться, что количество df1, df2 и df3 равно (хотя это работает только в том случае, если нет повторяющихся строк, если есть разные строки дубликатов, это метод все равно может вернуть true).

Другим вариантом будет получение базовых RDD обоих DataFrames, сопоставление с (Row, 1), выполнение reduceByKey для подсчета числа каждого Row и последующее объединение двух полученных RDD, а затем выполнение регулярного агрегата и return false, если какой-либо из итераторов не равен.

Ответ 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

Этот подход правильно обрабатывает случаи, когда в DataFrames могут быть дублированные строки, строки в разных порядках и/или столбцы в разных порядках.

Например:

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(), что, вероятно, ускорит процесс.

Ответ 3

Библиотека spark-fast-tests имеет два метода для сравнения DataFrame (я являюсь создателем библиотеки):

Метод assertSmallDataFrameEquality собирает DataFrames в драйвере node и делает сравнение

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, и я нашел его достаточным для своих тестовых наборов.

Ответ 4

Java:

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

Ответ 5

Вы можете сделать это, используя небольшую дедупликацию в сочетании с полным внешним соединением. Преимущество такого подхода заключается в том, что он не требует, чтобы вы собирали результаты для драйвера и избегали запуска нескольких заданий.

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()

Ответ 6

Попробуйте сделать следующее:

df1.except(df2).isEmpty

Ответ 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);}

метод столбцов является дополнительным и может быть заменен любым методом, который возвращает Seq

Логика:

1. Объединение обоих наборов данных, если столбцы не совпадают, будет выдано исключение и, следовательно, вернет false.
2. Если столбцы совпадают, то groupBy для всех столбцов и добавить количество столбцов. Теперь все строки имеют количество, кратное 2 (даже для повторяющихся строк).
3. Проверьте, есть ли какая-либо строка с числом, не делимым на 2, это дополнительные строки.