Spark DataFrame: does groupBy after orderBy maintain that order?

by

groupBy after orderBy doesn’t maintain order, as others have pointed out. What you want to do is use a Window function, partitioned on id and ordered by hours. You can collect_list over this and then take the max (largest) of the resulting lists since they go cumulatively (i.e. the first hour will only have itself in the list, the second hour will have 2 elements in the list, and so on).

Complete example code:

import org.apache.spark.sql.functions._
import org.apache.spark.sql.expressions.Window
import spark.implicits._

val data = Seq(
    ( "id1", 0, 12),
    ("id1", 1, 55),
    ("id1", 23, 44),
    ("id2", 0, 12),
    ("id2", 1, 89),
    ("id2", 23, 34)
).toDF("id", "hour", "count")

val mergeList = udf{(strings: Seq[String]) => strings.mkString(":")}

data.withColumn(
    "collected",
    collect_list($"count").over(
        Window.partitionBy("id").orderBy("hour")
    )
)
.groupBy("id")
.agg(max($"collected").as("collected"))
.withColumn("hourly_count", mergeList($"collected"))
.select("id", "hourly_count")
.show

This keeps us within the DataFrame world. I also simplified the UDF code the OP was using.

Output:

+---+------------+
| id|hourly_count|
+---+------------+
|id1|    12:55:44|
|id2|    12:89:34|
+---+------------+

More Related Contents:

  1. How to calculate the size of dataframe in bytes in Spark?
  2. How to convert rdd object to dataframe in spark
  3. How to define and use a User-Defined Aggregate Function in Spark SQL?
  4. Automatically and Elegantly flatten DataFrame in Spark SQL
  5. Spark extracting values from a Row
  6. How to avoid duplicate columns after join?
  7. Flattening Rows in Spark
  8. Caused by: java.lang.NullPointerException at org.apache.spark.sql.Dataset
  9. Provide schema while reading csv file as a dataframe
  10. Defining a UDF that accepts an Array of objects in a Spark DataFrame?
  11. Processing multiple files as independent RDD’s in parallel
  12. How to flatmap a nested Dataframe in Spark
  13. MatchError while accessing vector column in Spark 2.0
  14. How to save DataFrame directly to Hive?
  15. Spark / Scala: forward fill with last observation
  16. About how to add a new column to an existing DataFrame with random values in Scala
  17. Joining Spark dataframes on the key
  18. How to use Column.isin with list?
  19. How to write unit tests in Spark 2.0+?
  20. how to filter out a null value from spark dataframe
  21. How to compare two dataframe and print columns that are different in scala
  22. What are the various join types in Spark?
  23. Renaming column names of a DataFrame in Spark Scala
  24. Replace missing values with mean – Spark Dataframe
  25. What is going wrong with `unionAll` of Spark `DataFrame`?
  26. Exploding nested Struct in Spark dataframe
  27. How to get keys and values from MapType column in SparkSQL DataFrame
  28. Spark-Monotonically increasing id not working as expected in dataframe?
  29. Difference between two rows in Spark dataframe
  30. Filter spark DataFrame on string contains

Leave a Comment