我有一条小溪

+------+-------------------+------+
|group |               time| label|
+------+-------------------+------+
|     a|2020-01-01 10:49:00|red   |
|     a|2020-01-01 10:51:00|yellow|
|     a|2020-01-01 12:49:00|blue  |
|     b|2020-01-01 12:44:00|red   |
|     b|2020-01-01 12:46:00|blue  |
|     c|2020-01-01 12:46:00|green |
+------+-------------------+------+

我想使用spark流来为每个组保留最近的时间。
对于sparkDataframe，我将使用窗口函数作为

val window = {
    Window
    .partitionBy("group")
    .orderBy($"time".desc)
}

df
.withColumn("rn",row_number.over(window))
.filter("rn = 1")
.drop("rn")
.show()

或者

df
.orderBy($"time".desc)
.dropDuplicates("group")

在spark streaming中执行相同操作的最佳方法是什么？如何以只存储最新解决方案的方式保存结果？
更新：我试图保持每个组只有一行与最近的时间。有没有可能将有状态转换用于 mapGroupsWithState 为了这个目的？