Когда именно водяные знаки сбрасывают запоздалые данные?

Question

Вот простой код, который я использую для проверки водяных знаков:

spark.readStream
  .textFile("C:\\Users\\Pavel_Orekhov\\Desktop\\stream")
  .map(_.split(","))
  .map(a => (a(0), a(1), a(2)))
  .toDF("hour", "hashTag", "userId")
  .selectExpr("CAST(hour as TimeStamp)","hashTag", "userId")
  .withWatermark("hour", "1 hour")
  .groupBy(
    window($"hour", "1 hour", "1 hour"),
    $"hashTag",
    $"userId"
  ).count()
  .writeStream
  .outputMode(OutputMode.Update())
  .format("console")
  .start().processAllAvailable()

Папка stream содержит один файл со следующим содержимым:

1994-12-28T09:03,pavel,123
1994-12-28T09:03,pavel,123
1994-12-28T09:03,pavel,123
1994-12-28T09:03,pavel,123
1994-12-28T09:03,pavel,123
1994-12-28T09:03,pavel,123
1994-12-28T10:03,pavel,123
1994-12-28T10:03,pavel,123
1994-12-28T11:03,pavel,123
1994-12-28T11:03,pavel,123
1994-12-28T09:03,pavel,123
1994-12-28T06:03,pavel,123

Вывод, который я получаю,это:

+--------------------+-------+------+-----+
|              window|hashTag|userId|count|
+--------------------+-------+------+-----+
|[1994-12-28 09:00...|  pavel|   123|    7|
|[1994-12-28 06:00...|  pavel|   123|    1|
|[1994-12-28 11:00...|  pavel|   123|    2|
|[1994-12-28 10:00...|  pavel|   123|    2|
+--------------------+-------+------+-----+

В текстовом файле, который я прочитал, вы можете видеть, что есть запись 9 утра и запись 6 утра, которая идет после записи 11 утра.Я думал, что они будут отброшены, потому что водяные знаки должны обновлять только те данные, которые мы получили в течение последнего часа.

Итак, почему они не удаляются?

Answer 1

Оказывается, это потому, что это всего лишь одна партия и данные в ней неупорядочены. Когда я создал новый файл со значением 1994-12-28T06:03,pavel,123, он был удален, потому что он является частью нового пакета.