apache-spark-sql - Spark SQL可以在GROUP BY聚合中使用FIRST_VALUE和LAST_VALUE(但这不是标准的)

Question

(在Spark 2.2和2.3上测试)

我正在使用Spark将股票交易报价汇总到每日OHLC(开-高-低-收盘)记录中。

输入数据就像

val data = Seq(("2018-07-11 09:01:00", 34.0), ("2018-07-11 09:04:00", 32.0), ("2018-07-11 09:02:00", 35.0), ("2018-07-11 09:03:00", 30.0), ("2018-07-11 09:00:00", 33.0), ("2018-07-12 09:01:00", 56.0), ("2018-07-12 09:04:00", 54.0), ("2018-07-12 09:02:00", 51.0), ("2018-07-12 09:03:00", 50.0), ("2018-07-12 09:00:00", 51.0)).toDF("time", "price")

data.createOrReplaceTempView("ticks")

data.show

scala>

显示为

+-------------------+-----+
|               time|price|
+-------------------+-----+
|2018-07-11 09:01:00| 34.0|
|2018-07-11 09:04:00| 32.0|
|2018-07-11 09:02:00| 35.0|
|2018-07-11 09:03:00| 30.0|
|2018-07-11 09:00:00| 33.0|
|2018-07-12 09:01:00| 56.0|
|2018-07-12 09:04:00| 54.0|
|2018-07-12 09:02:00| 51.0|
|2018-07-12 09:03:00| 50.0|
|2018-07-12 09:00:00| 51.0|
+-------------------+-----+

所需的输出是

+----------+----+----+----+-----+
|      date|open|high| low|close|
+----------+----+----+----+-----+
|2018-07-11|33.0|35.0|30.0| 32.0|
|2018-07-12|51.0|56.0|50.0| 54.0|
+----------+----+----+----+-----+

已经有许多SQL解决方案，例如 this和 this。

由于SQL的限制，这些解决方案很麻烦。

今天，我发现Spark SQL可以在 FIRST_VALUE聚合 which is not allowed in standard SQL中使用 LAST_VALUE和 GROUP BY。

Spark SQL的这种局限性得出了一个整洁的解决方案。

SELECT
    TO_DATE(time) AS date,
    FIRST_VALUE(price) AS open,
    MAX(price) AS high,
    MIN(price) AS low,
    LAST_VALUE(price) AS close
FROM ticks
GROUP BY TO_DATE(time)

你可以试试看

spark.sql("SELECT TO_DATE(time) AS date, FIRST(price) AS open, MAX(price) AS high, MIN(price) AS low, LAST(price) AS close FROM ticks GROUP BY TO_DATE(time)").show

scala>

显示为

+----------+----+----+----+-----+
|      date|open|high| low|close|
+----------+----+----+----+-----+
|2018-07-11|34.0|35.0|30.0| 33.0|
|2018-07-12|56.0|56.0|50.0| 51.0|
+----------+----+----+----+-----+

但是，以上结果是不正确的。
FIRST_VALUE和 LAST_VALUE需要确定性排序才能获得确定性结果。

我可以通过在分组之前添加 orderBy来更正它。

import org.apache.spark.sql.functions._

data.orderBy("time").groupBy(expr("TO_DATE(time)").as("date")).agg(first("price").as("open"), max("price").as("high"), min("price").as("low"), last("price").as("close")).show

scala>

显示为

+----------+----+----+----+-----+
|      date|open|high| low|close|
+----------+----+----+----+-----+
|2018-07-11|33.0|35.0|30.0| 32.0|
|2018-07-12|51.0|56.0|50.0| 54.0|
+----------+----+----+----+-----+

这是正确的，期望的!

我的问题是，以上代码是否有效？是否可以保证此订购？我们可以在严肃的产品中使用此非标准功能吗？

这个问题的重点在于，在标准SQL中，我们只能先对 GROUP BY和 ORDER BY进行排序，然后对 ORDER BY和 GROUP BY进行排序。 GROUP BY将忽略 ORDER BY的顺序。

我还想知道Spark SQL是否可以按照所需的顺序执行这样的 GROUP BY，标准SQL还能为此发明一种语法吗？

P.S.

我可以想到一些依赖确定性排序的聚合函数。

WITH ORDER BY time SELECT COLLECT_LIST(price) GROUP BY stockID

WITH ORDER BY time SELECT SUM(SQUARE(price - LAG(price, 1, 0))) GROUP BY stockID

没有 WITH ORDER BY time，我们如何在标准SQL中对COLLECTed_LIST进行排序？

这些示例表明“按所需顺序排列的 GROUP BY”仍然有用。

Answer 1

不能保证按By/agg分组排序，可以将窗口功能与按键分区和按时间排序一起使用