DataFrame de l'égalité dans Apache Spark

Il y a certaines façons de Apache Spark suites de test, mais la plupart de ces consistent à collecter les données en local et si vous voulez faire de l'égalité des tests sur de grandes DataFrames alors ce n'est probablement pas une solution adaptée.

Vérifier le schéma d'abord, et puis vous pouvez faire une intersection à df3 et vérifier que le comte de df1,df2 & df3 sont tous égaux (cependant cela ne fonctionne que si il n'y a pas de doublons, si différents des doublons de lignes cette méthode pourrait toujours retourner true).

Une autre option serait d'obtenir le sous-jacent Rdd à la fois de la DataFrames, de la cartographie (Ligne, 1), faire un reduceByKey pour compter le nombre de chaque Ligne, puis cogrouping les deux résultant de la Rdd et puis faire régulièrement de l'agrégat et de retourner false si l'un des itérateurs ne sont pas égaux.

OriginalL'auteur Holden

La spark-rapide-tests bibliothèque a deux méthodes pour faire de l'DataFrame comparaisons (je suis le créateur de la bibliothèque):

La assertSmallDataFrameEquality méthode de collecte DataFrames sur le pilote nœud et rend la comparaison

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

La assertLargeDataFrameEquality méthode compare DataFrames réparties sur plusieurs machines (le code est essentiellement copié à partir de l'étincelle-essai-de 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 est plus rapide pour les petites DataFrame comparaisons et j'ai trouvé que c'est suffisant pour mon test suites.

OriginalL'auteur Powers

Vous pouvez faire cela en utilisant un peu de déduplication en combinaison avec une jointure externe complète. L'avantage de cette approche est qu'il ne vous oblige pas à recueillir les résultats pour le pilote, et qu'il évite l'exécution de plusieurs emplois.

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

OriginalL'auteur Herman van Hovell