Spark3.0 Structured Streaming时间字段排序窗口报非时间窗口错误原因咨询

• 第一，你使用的是批处理场景的通用开窗函数，且采用了默认的行边界窗口框架，而非Spark Structured Streaming支持的时间窗口类型。Spark流处理面向无界数据流，普通Window.partitionBy().orderBy()定义的开窗如果没有明确的时间边界，需要保留同key下的所有历史数据才能完成计算，不符合流处理的资源约束，所以无论排序字段是否为时间类型，这类开窗都不被支持。你看到的报错中Non-time-based windows不是指排序字段非时间类型，而是指开窗本身不是流处理规定的、带固定时间区间的滚动/滑动/会话窗口，或没有配置基于时间的范围窗口框架。
• 第二，水印配置位置和关联字段错误。你把withWatermark放在了开窗操作之后，Spark在解析开窗逻辑时还感知不到水印的时间约束；同时你水印绑定的是datetime字段，但开窗排序使用的是独立生成的time字段，二者没有关联，Spark无法确认time字段的数据过期规则，也无法推导开窗的计算边界。
• 第三，你通过current_timestamp()生成的是处理时间字段，而非事件时间，这类时间会随每次计算实时变化，不适合用于流数据的排序和去重逻辑，会导致计算结果不可重复。

你当前的逻辑是取每个key下最早到达的一条数据，可以调整为以下实现：

// 第一步：生成时间字段并先配置水印，水印绑定排序用的时间字段
controlDataFrame = controlDataFrame
  .withColumn("Make Coffee", $"value")
  .withColumn("time", current_timestamp())
  .withWatermark("time", "10 seconds") // 水印加在排序用的time字段上，且放在聚合前
  // 用分组+时间窗口替代通用开窗，10秒窗口和你的水印阈值匹配
  .groupBy($"key", window($"time", "10 seconds"))
  .agg(
    first("value").alias("value"),
    first("Make Coffee").alias("Make Coffee"),
    first("time").alias("time")
  )
  .withColumn("digitalTwinId", lit(digitalTwinId))

如果你的需求是按key去重，保留最早的一条，也可以直接使用dropDuplicatesWithinWatermark算子简化实现：

controlDataFrame = controlDataFrame
  .withColumn("Make Coffee", $"value")
  .withColumn("time", current_timestamp())
  .withWatermark("time", "10 seconds")
  .dropDuplicatesWithinWatermark("key")
  .withColumn("digitalTwinId", lit(digitalTwinId))

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