GroupBy和Aggregation不保留Spark SQL排序顺序？答案

【问题标题】：Spark SQL Sort order not retained by GroupBy and Aggregation?GroupBy和Aggregation不保留Spark SQL排序顺序？
【发布时间】：2017-06-01 00:23:41
【问题描述】：

我使用 Spark 2.1。

如果我运行以下示例：

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

上面代码的输出是：

+---+------------------+
| 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|
+---+------------------+

目的是获取与每个 id 的最新日期相关的分数：

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

但这显然没有奏效，因为结果是不确定的。我们是否必须使用窗口函数来实现这一点？

【问题讨论】：

你能更新你的预期输出吗？
你有什么版本的spark？
尝试使用 spark 2.1 重新生成问题。获得一致的结果。您能否在执行显示操作之前尝试“dfAgg.cache”，如果不一致仍然存在，请告诉我？
我仔细检查了 - 当我第一次运行它时，我得到了预期的结果 5 次。再跑一次，它变了！按照建议尝试 .cache 似乎给出了一致的结果，尽管在我一直运行它的情况下，当返回的分数为 1 时，“错误”的答案

标签： scala apache-spark apache-spark-sql

【解决方案1】：

查看 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.).
 */

表明不幸的是这是预期的行为。

所以在回答我的问题时，目前看来像 SPARK DataFrame: select the first row of each group 所述的 Window 函数可能是最好的方法。

【讨论】：

【解决方案2】：

您可以尝试使用 orderBy 而不是 sort，即使 javadoc 说它们是相同的

/** * 返回按给定表达式排序的新数据集。 * 这是sort 函数的别名。 * * @group typedrel * @since 2.0.0 / @scala.annotation.varargs def orderBy(sortCol: String, sortCols: String): Dataset[T] = sort(sortCol, sortCols : _*)

你可以试试

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

【讨论】：