Spark SQL Sort order not retained by GroupBy and Aggregation?

Question

I use Spark 2.1.

If I run the following example:

val seq = Seq((123,"2016-01-01","1"),(123,"2016-01-02","2"),(123,"2016-01-03","3"))

val df = seq.toDF("id","date","score")

val dfAgg = df.sort("id","date").groupBy("id").agg(last("score"))

dfAgg.show
dfAgg.show
dfAgg.show
dfAgg.show
dfAgg.show

The output of above code is:

+---+------------------+
| id|last(score, false)|
+---+------------------+
|123|                 1|
+---+------------------+

+---+------------------+
| id|last(score, false)|
+---+------------------+
|123|                 2|
+---+------------------+

+---+------------------+
| id|last(score, false)|
+---+------------------+
|123|                 1|
+---+------------------+

+---+------------------+
| id|last(score, false)|
+---+------------------+
|123|                 3|
+---+------------------+

+---+------------------+
| id|last(score, false)|
+---+------------------+
|123|                 3|
+---+------------------+

The intention was to get the score associated with the latest date for each id:

+---+------------------+
| id|last(score, false)|
+---+------------------+
|123|                 3|
+---+------------------+

but this clearly hasn't worked as the result is non-deterministic. Do we have to use window functions to achieve this?

tried to regenerate the issue with spark 2.1..getting consistent result. Could you please try "dfAgg.cache" before performing show operation and let me know if inconsistency is still there? — m-bhole, May 30 '17 at 16:56
I double checked - when I ran it the first time I got the expected result 5 times. Ran it once more and it varied! Trying the .cache as suggested seems to give a consistent result, though in the cases which I've run it consistently the 'wrong' answer when the score returned is 1 — user2682459, May 30 '17 at 17:08

user2682459 · Accepted Answer · 2017-05-30T21:42:12.817

Looking at the docs for org.apache.spark.sql.catalyst.expressions.aggregate.Last:

/**
 * Returns the last value of `child` for a group of rows. If the last value of `child`
 * is `null`, it returns `null` (respecting nulls). Even if [[Last]] is used on an already
 * sorted column, if we do partial aggregation and final aggregation (when mergeExpression
 * is used) its result will not be deterministic (unless the input table is sorted and has
 * a single partition, and we use a single reducer to do the aggregation.).
 */

shows that unfortunately this is expected behaviour.

So in answer to my question, for now it seems like Window functions, as described SPARK DataFrame: select the first row of each group may be the best way forwards.

score -3 · Answer 2 · answered May 31 '17 at 16:23

You can try with orderBy instead of sort even though the javadoc says they are same

/** * Returns a new Dataset sorted by the given expressions. * This is an alias of the sort function. * * @group typedrel * @since 2.0.0 / @scala.annotation.varargs def orderBy(sortCol: String, sortCols: String): Dataset[T] = sort(sortCol, sortCols : _*)

You can try

val dfAgg = df.orderBy("id","date").groupBy("id").agg(last("score"))

Spark SQL Sort order not retained by GroupBy and Aggregation?

2 Answers2