Spark Structured Streaming水印失效：窗口聚合仍处理历史数据

核心原因 #

1. 输出模式限制：你使用了complete输出模式，该模式要求Spark输出所有聚合结果，因此水印的状态清理机制完全不生效——Spark会永久保留所有窗口的聚合状态，无论数据多旧。
2. 事件时间列类型错误：如果你的dateTime字段在schema中是String类型而非Timestamp类型，水印无法正确识别事件时间，自然无法触发旧数据的过滤逻辑。
3. 滑动窗口的正常行为：你配置了10分钟窗口、5分钟滑动步长，单条数据会匹配两个重叠窗口，这是滑动窗口的预期行为，但旧数据的窗口本应被水印清理。

修复步骤 #

1. 修正事件时间列类型 #

确保dateTime被解析为Timestamp类型。如果schema中该字段是字符串，先转换：

import org.apache.spark.sql.functions.to_timestamp

val jsonDF = spark.readStream.format("json").schema(schema).load("data-source")
// 转换字符串为Timestamp类型，匹配你的时间格式
val timestampDF = jsonDF.withColumn("dateTime", to_timestamp($"dateTime", "yyyy-MM-dd'T'HH:mm:ss"))

2. 更换输出模式 #

将输出模式改为update或append，这两种模式才会触发水印的状态清理：

• update模式：仅输出有更新的聚合结果（适合需要追踪所有变化的场景）
• append模式：仅输出窗口已关闭且不会再收到数据的聚合结果（需确保水印延迟足够覆盖数据乱序）

示例代码（改用update模式）：

val result = timestampDF
  .withWatermark("dateTime", "10 minutes")
  .groupBy(window($"dateTime", "10 minutes", "5 minutes"), $"location")
  .sum("value")

val query = result.writeStream
  .outputMode("update") // 替换为update或append
  .format("console")
  .queryName("location-query")
  .start()

3. 验证效果 #

重新启动查询后，旧数据（如2022-12-01的记录）对应的窗口会在水印超时后被清理，后续输出只会包含符合时间范围的窗口聚合结果。

额外说明 #

若必须使用complete模式，Spark无法自动清理旧状态，你需要手动实现窗口过滤逻辑，比如在聚合后添加条件过滤窗口结束时间大于当前事件时间减去水印延迟：

import org.apache.spark.sql.functions.current_timestamp

val filteredResult = result.filter(
  $"window.end" >= current_timestamp().minusMinutes(10)
)

但这种方式无法自动清理状态，会导致状态持续膨胀，不建议长期使用。

内容的提问来源于stack exchange，提问作者beatrice