How to use groupBy to collect rows into a map?

by

Following will work with Spark 2.0. You can use map function available since 2.0 release to get columns as Map.

val df1 = df.groupBy(col("school_name")).agg(collect_list(map($"name",$"age")) as "map")
df1.show(false)

This will give you below output.

+-----------+------------------------------------+
|school_name|map                                 |
+-----------+------------------------------------+
|school B   |[Map(cathy -> 10), Map(shaun -> 5)] |
|school A   |[Map(michael -> 7), Map(emily -> 5)]|
+-----------+------------------------------------+

Now you can use UDF to join individual Maps into single Map like below.

import org.apache.spark.sql.functions.udf
val joinMap = udf { values: Seq[Map[String,Int]] => values.flatten.toMap }

val df2 = df1.withColumn("map", joinMap(col("map")))
df2.show(false)

This will give required output with Map[String,Int].

+-----------+-----------------------------+
|school_name|map                          |
+-----------+-----------------------------+
|school B   |Map(cathy -> 10, shaun -> 5) |
|school A   |Map(michael -> 7, emily -> 5)|
+-----------+-----------------------------+

If you want to convert a column value into JSON String then Spark 2.1.0 has introduced to_json function.

val df3 = df2.withColumn("map",to_json(struct($"map")))
df3.show(false)

The to_json function will return following output.

+-----------+-------------------------------+
|school_name|map                            |
+-----------+-------------------------------+
|school B   |{"map":{"cathy":10,"shaun":5}} |
|school A   |{"map":{"michael":7,"emily":5}}|
+-----------+-------------------------------+

More Related Contents:

  1. Using a column value as a parameter to a spark DataFrame function
  2. Split Spark Dataframe string column into multiple columns
  3. DataFrame join optimization – Broadcast Hash Join
  4. Overwrite specific partitions in spark dataframe write method
  5. How to control partition size in Spark SQL
  6. Avoid performance impact of a single partition mode in Spark window functions
  7. How to import multiple csv files in a single load?
  8. Convert date from String to Date format in Dataframes
  9. What is the meaning of partitionColumn, lowerBound, upperBound, numPartitions parameters?
  10. Spark DataFrame: count distinct values of every column
  11. How to split a list to multiple columns in Pyspark?
  12. pyspark dataframe filter or include based on list
  13. Use collect_list and collect_set in Spark SQL
  14. What is the maximum size for a broadcast object in Spark?
  15. Why does format(“kafka”) fail with “Failed to find data source: kafka.” (even with uber-jar)?
  16. How to find mean of grouped Vector columns in Spark SQL?
  17. Generate a Spark StructType / Schema from a case class
  18. How to loop through each row of dataFrame in pyspark
  19. How does createOrReplaceTempView work in Spark?
  20. Fill in null with previously known good value with pyspark
  21. PySpark – get row number for each row in a group
  22. Keep only duplicates from a DataFrame regarding some field
  23. Using windowing functions in Spark
  24. How can I access python variable in Spark SQL?
  25. PySpark: How to fillna values in dataframe for specific columns?
  26. Why spark application fail with “executor.CoarseGrainedExecutorBackend: Driver Disassociated”?
  27. How to calculate Median in spark sqlContext for column of data type double
  28. Spark: disk I/O on stage boundaries explanation
  29. spark.ml StringIndexer throws ‘Unseen label’ on fit()
  30. PySpark error: AttributeError: ‘NoneType’ object has no attribute ‘_jvm’

Leave a Comment