Automatically and Elegantly flatten DataFrame in Spark SQL

by

The short answer is, there’s no “accepted” way to do this, but you can do it very elegantly with a recursive function that generates your select(...) statement by walking through the DataFrame.schema.

The recursive function should return an Array[Column]. Every time the function hits a StructType, it would call itself and append the returned Array[Column] to its own Array[Column].

Something like:

import org.apache.spark.sql.Column
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.functions.col

def flattenSchema(schema: StructType, prefix: String = null) : Array[Column] = {
  schema.fields.flatMap(f => {
    val colName = if (prefix == null) f.name else (prefix + "." + f.name)

    f.dataType match {
      case st: StructType => flattenSchema(st, colName)
      case _ => Array(col(colName))
    }
  })
}

You would then use it like this:

df.select(flattenSchema(df.schema):_*)

More Related Contents:

  1. How to query JSON data column using Spark DataFrames?
  2. Encoder error while trying to map dataframe row to updated row
  3. How do I detect if a Spark DataFrame has a column
  4. Spark Scala: How to convert Dataframe[vector] to DataFrame[f1:Double, …, fn: Double)]
  5. How to define schema for custom type in Spark SQL?
  6. How to avoid duplicate columns after join?
  7. Flattening Rows in Spark
  8. How can I change column types in Spark SQL’s DataFrame?
  9. Spark unionAll multiple dataframes
  10. How can I pass extra parameters to UDFs in Spark SQL?
  11. How to split a dataframe into dataframes with same column values?
  12. How to zip two (or more) DataFrame in Spark
  13. How to define a custom aggregation function to sum a column of Vectors?
  14. Spark DataFrame: does groupBy after orderBy maintain that order?
  15. Better way to convert a string field into timestamp in Spark
  16. Spark UDAF with ArrayType as bufferSchema performance issues
  17. Apache Spark how to append new column from list/array to Spark dataframe
  18. How to use COGROUP for large datasets
  19. Why does join fail with “java.util.concurrent.TimeoutException: Futures timed out after [300 seconds]”?
  20. how to filter out a null value from spark dataframe
  21. What are the various join types in Spark?
  22. Provide schema while reading csv file as a dataframe in Scala Spark
  23. Replace missing values with mean – Spark Dataframe
  24. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  25. How to calculate the size of dataframe in bytes in Spark?
  26. Spark-Monotonically increasing id not working as expected in dataframe?
  27. Calculate Cosine Similarity Spark Dataframe
  28. How to create a Dataset of Maps?
  29. Spark DataFrames when udf functions do not accept large enough input variables
  30. Filter spark DataFrame on string contains

Leave a Comment