为什么我的火花水印演示不起作用？

如何解决为什么我的火花水印演示不起作用？

我想运行带有水印的结构流，但似乎不起作用：

import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming._
val in=spark.readStream.schema("id INT,time TIMESTAMP").csv("data/sensor/*.txt").withWatermark("time","1 seconds")
in.createOrReplaceTempView("in")
val in2=spark.sql("select id,window,count(1),collect_list(concat(time)) from in group by id,window(time,'5 second') order by window,id")
val writer=in2.writeStream.format("console").option("truncate","false").outputMode("complete").trigger(Trigger.ProcessingTime("1 second")).option("checkpointLocation","chktmp5")
val stream=writer.start

我使用一个带有csv的文件夹作为流源，我会在特殊处理时间生成csv，这里是计划：

eventtime processtime
8:57:01         8:57:11
8:57:03         8:57:13
8:57:01         8:57:25

当它结束时，你可以看到所有的处理时间作为文件时间：

me@ubuntu:~/Downloads/spark-3.0.1-bin-hadoop2.7/bin/data/sensor$ ls --full-time
total 24
-rw-rw-r-- 1 yz yz  22 2021-03-25 08:57:11.811041590 -0700 1.txt
-rw-rw-r-- 1 yz yz  22 2021-03-25 08:57:13.811052616 -0700 2.txt
-rw-rw-r-- 1 yz yz  22 2021-03-25 08:57:25.815118132 -0700 3.txt

和 1.txt 包含：

3,2021-03-25 08:57:01

2.txt 包含：

2,2021-03-25 08:57:03

3.txt 包含：

3,2021-03-25 08:57:01

显然，事件时间为 08:57:01 处理时间为 08:57:25 的事件为时已晚，应该放弃，但实际上不是，因为我使用控制台接收器，我可以看到：

-------------------------------------------
Batch: 2
-------------------------------------------
+---+------------------------------------------+--------+------------------------------------------+
|id |window                                    |count(1)|collect_list(concat(CAST(time AS STRING)))|
+---+------------------------------------------+--------+------------------------------------------+
|2  |[2021-03-25 08:57:00,2021-03-25 08:57:05]|1       |[2021-03-25 08:57:03]                     |
|3  |[2021-03-25 08:57:00,2021-03-25 08:57:05]|2       |[2021-03-25 08:57:01,2021-03-25 08:57:01] |

+---+------------------------------------------+--------+------------------------------------------+

所以我觉得水印不起作用，为什么？