每个键触发聚合事件集，包括它们的更改时间戳

Question

对于以下数据框：

+----+--------+-------------------+----+
|user|      dt|         time_value|item|
+----+--------+-------------------+----+
| id1|20200101|2020-01-01 00:00:00|   A|
| id1|20200101|2020-01-01 10:00:00|   B|
| id1|20200101|2020-01-01 09:00:00|   A|
| id1|20200101|2020-01-01 11:00:00|   B|
+----+--------+-------------------+----+

我想捕获所有独特的项目，即collect_set，但保留自己的time_value

import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.functions.unix_timestamp
import org.apache.spark.sql.functions.collect_set
import org.apache.spark.sql.types.TimestampType
val timeFormat = "yyyy-MM-dd HH:mm"
val dx = Seq(("id1", "20200101", "2020-01-01 00:00", "A"), ("id1", "20200101","2020-01-01 10:00", "B"), ("id1", "20200101","2020-01-01 9:00", "A"), ("id1", "20200101","2020-01-01 11:00", "B")).toDF("user", "dt","time_value", "item").withColumn("time_value", unix_timestamp(col("time_value"), timeFormat).cast(TimestampType))
dx.show

一个

dx.groupBy("user", "dt").agg(collect_set("item")).show
+----+--------+-----------------+                                               
|user|      dt|collect_set(item)|
+----+--------+-----------------+
| id1|20200101|           [B, A]|
+----+--------+-----------------+

信号从A切换到B时不保留time_value信息。如何保留item中每组的时间值信息？

是否可以在窗口函数中使用 collect_set 以达到预期的效果？目前，我只能想到：

1. 使用窗口函数来确定事件对
2. 过滤以更改事件
3. 聚合

需要多次洗牌。或者，也可以使用 UDF (collect_list(sort_array(struct(time_value, item))))，但这似乎也很笨拙。

有没有更好的办法？

Answer 1

我确实会使用窗口函数来隔离更改点，我认为没有其他选择：

val win = Window.partitionBy($"user",$"dt").orderBy($"time_value")

dx
.orderBy($"time_value")
.withColumn("item_change_post",coalesce((lag($"item",1).over(win)=!=$"item"),lit(false)))
.withColumn("item_change_pre",lead($"item_change_post",1).over(win))
.where($"item_change_pre" or $"item_change_post")
.show()

+----+--------+-------------------+----+----------------+---------------+
|user|      dt|         time_value|item|item_change_post|item_change_pre|
+----+--------+-------------------+----+----------------+---------------+
| id1|20200101|2020-01-01 09:00:00|   A|           false|           true|
| id1|20200101|2020-01-01 10:00:00|   B|            true|          false|
+----+--------+-------------------+----+----------------+---------------+

然后使用类似groupBy($"user",$"dt").agg(collect_list(struct($"time_value",$"item")))

我认为不会发生多次随机播放，因为您总是按相同的键进行分区/分组。

您可以尝试通过将初始数据帧聚合到每个 item 的最小/最大 time_value 来提高效率，然后执行与上述相同的操作。