Spark groupby, sort values, then take first and last

1
votes

I'm using Apache Spark and have a dataframe that looks like this:

scala> df.printSchema
root
 |-- id: string (nullable = true)
 |-- epoch: long (nullable = true)


scala> df.show(10)
+--------------------+-------------+
|                 id |        epoch|
+--------------------+-------------+
|6825a28d-abe5-4b9...|1533926790847|
|6825a28d-abe5-4b9...|1533926790847|
|6825a28d-abe5-4b9...|1533180241049|
|6825a28d-abe5-4b9...|1533926790847|
|6825a28d-abe5-4b9...|1532977853736|
|6825a28d-abe5-4b9...|1532531733106|
|1eb5f3a4-a68c-4af...|1535383198000|
|1eb5f3a4-a68c-4af...|1535129922000|
|1eb5f3a4-a68c-4af...|1534876240000|
|1eb5f3a4-a68c-4af...|1533840537000|
+--------------------+-------------+
only showing top 10 rows

I want to group by the id field to get all the epoch timestamps together for an id. I then want to sort the epochs by ascending timestamp and then take the first and last epochs.

I used the following query, but the first and last epoch values appear to be taken in the order that they appear in the original dataframe. I want the first and last to be taken from a sorted ascending order.

scala> val df2 = df2.groupBy("id").
                 agg(first("epoch").as("first"), last("epoch").as("last"))

scala> df2.show()
+--------------------+-------------+-------------+                              
|                  id|        first|         last|
+--------------------+-------------+-------------+
|4f433f46-37e8-412...|1535342400000|1531281600000|
|d0cba2f9-cc04-42c...|1535537741000|1530448494000|
|6825a28d-abe5-4b9...|1533926790847|1532531733106|
|e963f265-809c-425...|1534996800000|1534996800000|
|1eb5f3a4-a68c-4af...|1535383198000|1530985221000|
|2e65a033-85ed-4e4...|1535660873000|1530494913413|
|90b94bb0-740c-42c...|1533960000000|1531108800000|
+--------------------+-------------+-------------+

How do I retrieve the first and last from the epoch list sorted by ascending epoch?

2
apache-sparkapache-spark-sql
Use min and max? – Psidom
I will later use string values, not just the numeric epoch. Will min and max also work for strings? – stackoverflowuser2010
I suppose so. Otherwise you need window functions. – Psidom
Solution with window stackoverflow.com/a/45210121/1465609 – wind

2 Answers

4
votes

first and last functions are meaningless when applied outside Window context. The value which is taken is purely arbitrary.

Instead you should

  • Use min / max functions if the logic conforms to basic ordering rules (alphanumeric for strings, arrays, and structs, numeric for numbers).

  • Strongly typed dataset with map -> groupByKey -> reduceGroups or groupByKey -> mapGroups otherwise.

-2
votes

You can just use min and max and cast the resulting columns to string. Here is one way to do it

   import org.apache.spark.sql.functions._
val df = Seq(("6825a28d-abe5-4b9",1533926790847.0),
("6825a28d-abe5-4b9",1533926790847.0),
("6825a28d-abe5-4b9",1533180241049.0),
("6825a28d-abe5-4b9",1533926790847.0),
("6825a28d-abe5-4b9",1532977853736.0),
("6825a28d-abe5-4b9",1532531733106.0),
("1eb5f3a4-a68c-4af",1535383198000.0),
("1eb5f3a4-a68c-4af",1535129922000.0),
("1eb5f3a4-a68c-4af",1534876240000.0),
("1eb5f3a4-a68c-4af",1533840537000.0)).toDF("id","epoch").withColumn("epoch",($"epoch"/1000.0).cast("timestamp"))

    +-----------------+--------------------+
|               id|               epoch|
+-----------------+--------------------+
|6825a28d-abe5-4b9|2018-08-10 18:46:...|
|6825a28d-abe5-4b9|2018-08-10 18:46:...|
|6825a28d-abe5-4b9|2018-08-02 03:24:...|
|6825a28d-abe5-4b9|2018-08-10 18:46:...|
|6825a28d-abe5-4b9|2018-07-30 19:10:...|
|6825a28d-abe5-4b9|2018-07-25 15:15:...|
|1eb5f3a4-a68c-4af| 2018-08-27 15:19:58|
|1eb5f3a4-a68c-4af| 2018-08-24 16:58:42|
|1eb5f3a4-a68c-4af| 2018-08-21 18:30:40|
|1eb5f3a4-a68c-4af| 2018-08-09 18:48:57|
+-----------------+--------------------+

    val df1 = df.groupBy("id").agg(min($"epoch").cast("string").as("first"), max($"epoch").cast("string"). as("last"))
df1.show

    +-----------------+--------------------+--------------------+
|               id|               first|                last|
+-----------------+--------------------+--------------------+
|6825a28d-abe5-4b9|2018-07-25 15:15:...|2018-08-10 18:46:...|
|1eb5f3a4-a68c-4af| 2018-08-09 18:48:57| 2018-08-27 15:19:58|
+-----------------+--------------------+--------------------+


    df1: org.apache.spark.sql.DataFrame = [id: string, first: string ... 1 more field]