我有以下Dataframe：

我需要对总线ID进行分组，并通过对spark结构化流中的时间进行排序来打印输出。
我在append模式下编写了以下代码：

solutionSDF = mapSDF.withWatermark("my_time", "0 seconds") \
        .groupBy(F.window("my_time", my_frequency, my_frequency), F.col("bus_id"),F.col("stop")) \
        .agg(F.sort_arrayF.collect_set("time"))

但这会打乱停站和时间的组合。
o/p应为：

+---------+-------------------+----+
|vehicleID|               time|stop|
+---------+-------------------+----+
|    30000|2013-01-10 09:01:00| 279|
|    30000|2013-01-10 09:04:00|1935|
|    30000|2013-01-10 09:15:00| 500|
|    30000|2013-01-10 09:25:00| 600|
|    20000|2013-01-10 09:03:00|1935|
|    20000|2013-01-10 09:10:00| 200|
|    20000|2013-01-10 09:20:00| 300|
+---------+-------------------+----+

有人能建议吗？提前谢谢