QBoard » Big Data » Big Data - Spark » DataFrame equality in Apache Spark

DataFrame equality in Apache Spark

  • Assume df1 and df2 are two DataFrames in Apache Spark, computed using two different mechanisms, e.g., Spark SQL vs. the Scala/Java/Python API.

    Is there an idiomatic way to determine whether the two data frames are equivalent (equal, isomorphic), where equivalence is determined by the data (column names and column values for each row) being identical save for the ordering of rows & columns?

    The motivation for the question is that there are often many ways to compute some big data result, each with its own trade-offs. As one explores these trade-offs, it is important to maintain correctness and hence the need to check for the equivalence/equality on a meaningful test data set.
      October 30, 2021 1:35 PM IST
    0
  • Java:

    assert resultDs.union(answerDs).distinct().count() == resultDs.intersect(answerDs).count();
      December 9, 2021 12:31 PM IST
    0
  • There are some standard ways in the Apache Spark test suites, however most of these involve collecting the data locally and if you want to do equality testing on large DataFrames then that is likely not a suitable solution.

    Checking the schema first and then you could do an intersection to df3 and verify that the count of df1,df2 & df3 are all equal (however this only works if there aren't duplicate rows, if there are different duplicates rows this method could still return true).

    Another option would be getting the underlying RDDs of both of the DataFrames, mapping to (Row, 1), doing a reduceByKey to count the number of each Row, and then cogrouping the two resulting RDDs and then do a regular aggregate and return false if any of the iterators are not equal.

      November 15, 2021 12:33 PM IST
    0
  • I don't know about idiomatic, but I think you can get a robust way to compare DataFrames as you describe as follows. (I'm using PySpark for illustration, but the approach carries across languages.)

    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

     

    This approach correctly handles cases where the DataFrames may have duplicate rows, rows in different orders, and/or columns in different orders.

    For example:

    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

     

    This approach is quite expensive, but most of the expense is unavoidable given the need to perform a full diff. And this should scale fine as it doesn't require collecting anything locally. If you relax the constraint that the comparison should account for duplicate rows, then you can drop the groupBy() and just do the subtract(), which would probably speed things up notably.

     
      February 2, 2022 1:49 PM IST
    0