Flattening Rows in Spark

by

You can use explode function:

scala> import org.apache.spark.sql.functions.explode
import org.apache.spark.sql.functions.explode


scala> val test = sqlContext.read.json(sc.parallelize(Seq("""{"a":1,"b":[2,3]}""")))
test: org.apache.spark.sql.DataFrame = [a: bigint, b: array<bigint>]

scala> test.printSchema
root
 |-- a: long (nullable = true)
 |-- b: array (nullable = true)
 |    |-- element: long (containsNull = true)

scala> val flattened = test.withColumn("b", explode($"b"))
flattened: org.apache.spark.sql.DataFrame = [a: bigint, b: bigint]

scala> flattened.printSchema
root
 |-- a: long (nullable = true)
 |-- b: long (nullable = true)

scala> flattened.show
+---+---+
|  a|  b|
+---+---+
|  1|  2|
|  1|  3|
+---+---+

More Related Contents:

  1. Exploding nested Struct in Spark dataframe
  2. How to query JSON data column using Spark DataFrames?
  3. Encoder error while trying to map dataframe row to updated row
  4. How do I detect if a Spark DataFrame has a column
  5. How to define schema for custom type in Spark SQL?
  6. How to avoid duplicate columns after join?
  7. How can I change column types in Spark SQL’s DataFrame?
  8. Spark unionAll multiple dataframes
  9. How can I pass extra parameters to UDFs in Spark SQL?
  10. How to split a dataframe into dataframes with same column values?
  11. Dropping a nested column from Spark DataFrame
  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. Create new Dataframe with empty/null field values
  22. What are the various join types in Spark?
  23. Provide schema while reading csv file as a dataframe in Scala Spark
  24. Spark: Add column to dataframe conditionally
  25. Replace missing values with mean – Spark Dataframe
  26. Apache Spark, add an “CASE WHEN … ELSE …” calculated column to an existing DataFrame
  27. How to calculate the size of dataframe in bytes in Spark?
  28. Spark-Monotonically increasing id not working as expected in dataframe?
  29. Calculate Cosine Similarity Spark Dataframe
  30. Difference between two rows in Spark dataframe

Leave a Comment