Apache Flink Sliding Window计算均值无结果问题求助

1. Upsert-Kafka 连接器与窗口语义不兼容 #

你使用的upsert-kafka连接器针对主键更新流设计（支持插入、更新、删除操作），但Flink窗口计算是为追加式无界流打造的。Upsert流的变更日志语义（如重复主键更新）会干扰窗口的事件时间聚合逻辑，导致窗口无法正确触发输出。

解决办法：
如果业务不需要处理更新/删除，改用普通kafka连接器消费追加流：

final TableDescriptor filteredPhasesDurationsTableDescriptor = TableDescriptor.forConnector("kafka")
       .schema(Schema.newBuilder()
             .column("id_fascicolo", DataTypes.BIGINT().notNull())
             .column("nrg", DataTypes.STRING())
             .column("giudice", DataTypes.STRING())
             .column("oggetto", DataTypes.STRING())
             .column("codice_oggetto", DataTypes.STRING())
             .column("ufficio", DataTypes.STRING())
             .column("sezione", DataTypes.STRING())
             .column("fase", DataTypes.STRING().notNull())
             .column("durata", DataTypes.BIGINT())
             .column("data_inizio", DataTypes.TIMESTAMP_LTZ(3))
             .column("data_fine", DataTypes.TIMESTAMP_LTZ(3))
             .watermark("data_fine", "data_fine - INTERVAL '5' SECOND")
             .build())
       .option(KafkaConnectorOptions.TOPIC, List.of("sicid.processor.filtered-phases-durations"))
       .option(KafkaConnectorOptions.PROPS_BOOTSTRAP_SERVERS, KAFKA_HOST)
       .option(KafkaConnectorOptions.KEY_FORMAT, "json")
       .option(KafkaConnectorOptions.VALUE_FORMAT, "json")
       .option(KafkaConnectorOptions.CONSUMER_STARTUP_MODE, "earliest")
       .build();

2. Watermark 未推进到窗口触发阈值 #

Flink事件时间窗口的触发完全依赖Watermark，只有当Watermark超过窗口结束时间时，窗口才会执行计算并输出结果。

你的测试数据中：

• 最晚的data_fine是2023-05-15 02:00:00.000，按Watermark规则data_fine - INTERVAL '5' SECOND，能推进的Watermark最多到2023-05-15 01:59:55.000
• 对于结束时间为2023-05-15 02:00:00.000的窗口，Watermark未达到触发阈值，因此窗口不会输出。

解决办法：

• 调整测试数据：添加一条data_fine为2023-05-15 02:00:05.000的事件，此时Watermark会推进到2023-05-15 02:00:00.000，触发对应窗口计算
• 处理流空闲场景：本地运行时添加空闲超时配置，避免无新事件时Watermark停滞：

  tEnv.getConfig().set("table.exec.source.idle-timeout", "10000"); // 10秒空闲后自动推进Watermark
  

3. 窗口时间范围验证 #

可以在查询中添加窗口起止时间，直观确认测试数据是否覆盖目标窗口：

.select(
      $("giudice"),
      $("fase"),
      $("w").start().as("window_start"),
      $("w").end().as("window_end"),
      $("durata").avg().as("mediaMobileGiudicePerFase")
);

内容的提问来源于stack exchange，提问作者E. Marotti