OutputMode.Complete 的 Spark 结构化流式水印

Question

我写了一个简单的查询，它应该忽略创建位置 < 上次事件时间 - 5 秒的数据。但是这个查询不起作用。所有数据都打印出来了。

我也尝试使用 window 函数 window($"created", "10 seconds", "10 seconds")，但没有用。

val inputStream = new MemoryStream[(Timestamp, String)](1, spark.sqlContext)

val df = inputStream.toDS().toDF("created", "animal")

val query = df
  .withWatermark("created", "5 seconds")
  .groupBy($"animal")
  .count()
  .writeStream
  .format("console")
  .outputMode(OutputMode.Complete())
  .start()

Answer 1

您需要按如下信息进行更多分组：

val windowedCounts = words
    .withWatermark("timestamp", "10 minutes")
    .groupBy(
        window($"timestamp", "10 minutes", "5 minutes"),
        $"word")
    .count()

此外，来自手册：

输出模式必须是追加或更新。完整模式要求保留所有聚合数据，因此不能使用水印来删除中间状态。