在 Spark 2.1.1 中获取 window 的最后一个元素

Question

我有一个数据框，其中有子类别，并且想要每个子类别的最后一个元素。

val windowSpec = Window.partitionBy("name").orderBy("count")
sqlContext
    .createDataFrame(
      Seq[(String, Int)](
        ("A", 1),
        ("A", 2),
        ("A", 3),
        ("B", 10),
        ("B", 20),
        ("B", 30)
      ))
    .toDF("name", "count")
    .withColumn("firstCountOfName", first("count").over(windowSpec))
    .withColumn("lastCountOfName", last("count").over(windowSpec))
    .show()

returns我有点奇怪：

+----+-----+----------------+---------------+                                   
|name|count|firstCountOfName|lastCountOfName|
+----+-----+----------------+---------------+
|   B|   10|              10|             10|
|   B|   20|              10|             20|
|   B|   30|              10|             30|
|   A|    1|               1|              1|
|   A|    2|               1|              2|
|   A|    3|               1|              3|
+----+-----+----------------+---------------+

正如我们所见，返回的 first 值是正确计算的，但 last 不是，它始终是列的当前值。

有人有解决方案来做我想做的事吗？

Answer 1

根据问题 SPARK-20969，您应该能够通过为 window 定义足够的界限来获得预期的结果，如下所示。

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

val windowSpec = Window
  .partitionBy("name")
  .orderBy("count")
  .rowsBetween(Window.unboundedPreceding, Window.unboundedFollowing)

sqlContext
  .createDataFrame(
    Seq[(String, Int)](
      ("A", 1),
      ("A", 2),
      ("A", 3),
      ("B", 10),
      ("B", 20),
      ("B", 30)
    ))
  .toDF("name", "count")
  .withColumn("firstCountOfName", first("count").over(windowSpec))
  .withColumn("lastCountOfName", last("count").over(windowSpec))
  .show()

或者，如果您在计算第一个和最后一个列的同一列上排序，您可以将 min 和 max 更改为未排序的 window，然后它应该也能正常工作。

在 Spark 2.1.1 中获取 window 的最后一个元素

Get the last element of a window in Spark 2.1.1

scala

window-functions

apache-spark

apache-spark-sql