Schema comparison of two dataframes in scala

by

Based on @Derek Kaknes‘s answer, here’s the solution I came up with for comparing schemas, being concerned only about column name, datatype & nullability and indifferent to metadata

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types.{DataType, StructField}

def getCleanedSchema(df: DataFrame): Map[String, (DataType, Boolean)] = {
    df.schema.map { (structField: StructField) =>
      structField.name.toLowerCase -> (structField.dataType, structField.nullable)
    }.toMap
  }

// Compare relevant information
def getSchemaDifference(schema1: Map[String, (DataType, Boolean)],
                        schema2: Map[String, (DataType, Boolean)]
                       ): Map[String, (Option[(DataType, Boolean)], Option[(DataType, Boolean)])] = {
  (schema1.keys ++ schema2.keys).
    map(_.toLowerCase).
    toList.distinct.
    flatMap { (columnName: String) =>
      val schema1FieldOpt: Option[(DataType, Boolean)] = schema1.get(columnName)
      val schema2FieldOpt: Option[(DataType, Boolean)] = schema2.get(columnName)

      if (schema1FieldOpt == schema2FieldOpt) None
      else Some(columnName -> (schema1FieldOpt, schema2FieldOpt))
    }.toMap
}

  • getCleanedSchema method extracts information of interest – column datatype & nullability and returns a map of column name to tuple

  • getSchemaDifference method returns a map containing only those columns that differ in the two schemas. If a column is absent in one of the two schemas, then it’s corresponding properties would be None

More Related Contents:

  1. How to convert rdd object to dataframe in spark
  2. Spark 2.0 Dataset vs DataFrame
  3. Spark – load CSV file as DataFrame?
  4. How to define partitioning of DataFrame?
  5. Automatically and Elegantly flatten DataFrame in Spark SQL
  6. Spark extracting values from a Row
  7. SparkSQL: How to deal with null values in user defined function?
  8. How to aggregate values into collection after groupBy?
  9. Dropping a nested column from Spark DataFrame
  10. How to read records in JSON format from Kafka using Structured Streaming?
  11. Spark Dataframe :How to add a index Column : Aka Distributed Data Index
  12. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  13. Processing multiple files as independent RDD’s in parallel
  14. MatchError while accessing vector column in Spark 2.0
  15. Spark / Scala: forward fill with last observation
  16. How to use Column.isin with list?
  17. Filling gaps in timeseries Spark
  18. How to use COGROUP for large datasets
  19. What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]
  20. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  21. Create new Dataframe with empty/null field values
  22. Perform a typed join in Scala with Spark Datasets
  23. Apache Spark: Get number of records per partition
  24. DataFrame-ified zipWithIndex
  25. Derive multiple columns from a single column in a Spark DataFrame
  26. Spark: Add column to dataframe conditionally
  27. How to save a spark DataFrame as csv on disk?
  28. How to sort by column in descending order in Spark SQL?
  29. Why is join not possible after show operator?
  30. How to create DataFrame from Scala’s List of Iterables?

Leave a Comment