我曾为我的团队尝试过类似的场景。请在下面找到。

val df=Seq(("05:59","1","A"),("06:00","1","A"),("7:00","1","B"),("7:00","1","A"),("7:01","1","A"),("7:02","1","A")).toDF("Timestamp","id","Grp")
df.createOrReplaceTempView("df")
val df2=spark.sql("select *,lag(grp) OVER w as prev_grp,lead(grp) OVER w as next_grp  from df  WINDOW w AS ( ORDER BY Timestamp)")
df2.createOrReplaceTempView("df2")

spark.sql("""select id,collect_list(grp)  from (select *,
  SUM(CASE WHEN (grp=prev_grp and grp = next_grp) THEN 0  
  WHEN (grp=next_grp and grp != prev_grp) THEN 1 
  WHEN (grp=prev_grp and (grp != next_grp or next_grp is null) ) THEN 0 
  ELSE 1 END) OVER
  (ORDER BY Timestamp
   ROWS BETWEEN UNBOUNDED PRECEDING
            AND CURRENT ROW)+1 as EVENT_SEQ
from df2
ORDER BY Timestamp) s group by EVENT_SEQ,id""").show(false)

输入

df.show(false)
+---------+---+---+
|Timestamp|id |Grp|
+---------+---+---+
|05:59    |1  |A  |
|06:00    |1  |A  |
|7:00     |1  |B  |
|7:00     |1  |A  |
|7:01     |1  |A  |
|7:02     |1  |A  |
+---------+---+---+

输出：

+---+-----------------+
|id |collect_list(grp)|
+---+-----------------+
|1  |[A, A, A]        |
|1  |[B]              |
|1  |[A, A]           |
+---+-----------------+