Databricks DLT滑动窗口缺失最后一个时间窗口问题求助

我在Databricks DLT流水线中，需要计算某列过去24小时的滚动平均值，每小时更新一次。使用以下代码实现：

@dlt.table()
def gold():
    df = dlt.read_stream("silver_table")
    
    # Define window for 24 hours with 1-hour slide
    window_spec_24h = window("fetch_ts", "24 hours", "1 hour")

    df.withWatermark("fetch_ts", "10 minutes")
      .groupBy(df.Id, window_spec_24h)
      .agg(avg("foo").alias("average_foo_24h"))

    return df

问题：结果DataFrame中始终缺失最后一个窗口。例如输入数据的fetch_ts包含2024-02-23T18:54:00.000，但输出窗口仅到结束时间为2024-02-23T18:00:00.000的窗口，该最新数据被排除。下一次触发时，之前缺失的窗口会出现，但最新批次的最后一个窗口又会缺失。请问这是什么原因？是否为设计如此？能否让结果包含结束时间为2024-02-23T19:00:00.000的窗口以纳入最新数据？注：移除水印后可获取最新窗口，但因需将计算后的DataFrame与原表进行流流连接，不能移除水印。

问题原因与解决方案 #

原因分析 #

• 水印的延迟触发逻辑：Spark Structured Streaming中，窗口结果的输出必须等待水印时间超过窗口结束时间。你设置的水印是10分钟，意味着只有当流中出现fetch_ts晚于「窗口结束时间+10分钟」的数据时，该窗口的计算结果才会被输出。比如结束时间19:00的窗口，要等到有数据的fetch_ts ≥ 19:10时，才会输出该窗口的结果。
• 窗口时间对齐特性：你的窗口是1小时滑动、24小时窗口，结束时间为整点（如18:00、19:00）。当数据时间为18:54时，对应的19:00结束窗口还未满足「水印覆盖窗口结束时间+10分钟」的条件，因此不会被输出。

是否为设计如此 #

是的，这是Spark Structured Streaming的默认设计。该机制是为了避免乱序数据导致的计算误差——如果过早输出窗口结果，后续到达的属于该窗口的乱序数据无法被纳入，最终结果会不准确。

可行解决方案 #

在保留水印的前提下，可通过以下方式调整以获取最新窗口结果：

1. 缩小水印延迟时间：如果你的数据乱序程度极低（比如几乎没有超过1分钟的乱序），可以将水印从10分钟缩短至1分钟甚至更短，大幅降低窗口结果的输出延迟。例如：

   df.withWatermark("fetch_ts", "1 minute")
     .groupBy(df.Id, window_spec_24h)
     .agg(avg("foo").alias("average_foo_24h"))
   

2. 结合allowedLateness与更新模式：使用outputMode("update")配合allowedLateness参数，允许窗口在水印之后仍接收一定时间的乱序数据。这种方式下，最新窗口的结果会在数据到达时先输出一次，后续有乱序数据进来时会自动更新结果。
3. 调整窗口偏移量：如果业务允许，可给窗口设置偏移量，让窗口结束时间提前。例如将窗口定义为window("fetch_ts", "24 hours", "1 hour", "5 minutes")，窗口结束时间变为18:05、19:05等，让最新数据能更早进入满足水印条件的窗口。

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