使用Scala获取与Spark数据集中最新时间戳对应的行

Question

我对Spark和Scala比较陌生。我有一个数据帧，它有以下格式：

| Col1 | Col2 | Col3 | Col_4 | Col_5 | Col_TS                  | Col_7 | 

| 1234 | AAAA | 1111 | afsdf | ewqre | 1970-01-01 00:00:00.0   | false |
| 1234 | AAAA | 1111 | ewqrw | dafda | 2017-01-17 07:09:32.748 | true  |
| 1234 | AAAA | 1111 | dafsd | afwew | 2015-01-17 07:09:32.748 | false |
| 5678 | BBBB | 2222 | afsdf | qwerq | 1970-01-01 00:00:00.0   | true  |
| 5678 | BBBB | 2222 | bafva | qweqe | 2016-12-08 07:58:43.04  | false |
| 9101 | CCCC | 3333 | caxad | fsdaa | 1970-01-01 00:00:00.0   | false |

我需要做的是获取对应于最新时间戳的行。在上面的示例中，键是Col1、Col2和Col3。Col_TS表示时间戳，Col_7是一个布尔值，用于确定记录的有效性。我想做的是找到一种方法，根据密钥对这些记录进行分组，并保留具有最新时间戳的记录。

因此，上述数据框中的操作输出应为：

| Col1 | Col2 | Col3 | Col_4 | Col_5 | Col_TS                  | Col_7 | 

| 1234 | AAAA | 1111 | ewqrw | dafda | 2017-01-17 07:09:32.748 | true  |
| 5678 | BBBB | 2222 | bafva | qweqe | 2016-12-08 07:58:43.04  | false |
| 9101 | CCCC | 3333 | caxad | fsdaa | 1970-01-01 00:00:00.0   | false |

我想出了一个部分解决方案，但是这样我只能返回记录分组的列键的数据帧，而不是其他列。

df = df.groupBy("Col1","Col2","Col3").agg(max("Col_TS"))


| Col1 | Col2 | Col3 | max(Col_TS)             |

| 1234 | AAAA | 1111 | 2017-01-17 07:09:32.748 |
| 5678 | BBBB | 2222 | 2016-12-08 07:58:43.04  | 
| 9101 | CCCC | 3333 | 1970-01-01 00:00:00.0   | 

有人能帮我想出执行此操作的Scala代码吗？

Answer 1

一个选项是首先按列对数据帧进行排序，然后按列1、列2和列3分组，并从每个列中选取最后一项：

val val_columns = Seq("Col_4", "Col_5", "Col_TS", "Col_7").map(x => last(col(x)).alias(x))

(df.orderBy("Col_TS")
   .groupBy("Col1", "Col2", "Col3")
   .agg(val_columns.head, val_columns.tail: _*).show)

+----+----+----+-----+-----+--------------------+-----+
|Col1|Col2|Col3|Col_4|Col_5|              Col_TS|Col_7|
+----+----+----+-----+-----+--------------------+-----+
|1234|AAAA|1111|ewqrw|dafda|2017-01-17 07:09:...| true|
|9101|CCCC|3333|caxad|fsdaa|1970-01-01 00:00:...|false|
|5678|BBBB|2222|bafva|qweqe|2016-12-08 07:58:...|false|
+----+----+----+-----+-----+--------------------+-----+

Answer 2

您可以按如下方式使用窗口功能

import org.apache.spark.sql.functions._
val windowSpec = Window.partitionBy("Col1","Col2","Col3").orderBy(col("Col_TS").desc)

df.withColumn("maxTS", first("Col_TS").over(windowSpec))
.select("*").where(col("maxTS") === col("Col_TS"))
.drop("maxTS")
  .show(false)

您应该得到如下输出

+----+----+----+-----+-----+----------------------+-----+
|Col1|Col2|Col3|Col_4|Col_5|Col_TS                |Col_7|
+----+----+----+-----+-----+----------------------+-----+
|5678|BBBB|2222|bafva|qweqe|2016-12-0807:58:43.04 |false|
|1234|AAAA|1111|ewqrw|dafda|2017-01-1707:09:32.748|true |
|9101|CCCC|3333|caxad|fsdaa|1970-01-0100:00:00.0  |false|
+----+----+----+-----+-----+----------------------+-----+