Flink SQL窗口不报告最终结果

Question

我使用Flink SQL计算基于事件时间的窗口分析。在我的数据源每天晚上空闲之前，一切都正常工作，之后直到第二天数据再次开始流动时才产生最后一分钟的结果。

CREATE TABLE input
    id STRING,
    data BIGINT,
    rowtime TIMESTAMP(3) METADATA FROM 'timestamp',
    WATERMARK FOR rowtime AS rowtime - INTERVAL '1' SECOND
WITH (
    'connector' = 'kafka',
    'topic' = 'input',
    'properties.bootstrap.servers' = 'localhost:9092',
    'scan.startup.mode' = 'latest-offset',
    'format' = 'json'
)

SELECT ...
FROM
  (SELECT * FROM
     TABLE(TUMBLE(TABLE input, DESCRIPTOR(rowtime), INTERVAL '1' MINUTES)))
GROUP BY ..., window_start, window_end

我已尝试将＜code＞设置为table.exec.source。空闲超时，但没有帮助。我能做什么？

Answer 1

table.exec.source.idle-timeout（以及与DataStream API for Watermark Strategy一起使用的相应with Idlness构造）检测空闲的输入分区并防止它们阻碍整体水印的进度。但是，为了使整体水印前进，仍然必须在某个地方有一些输入。

一些选项:

（1）解决问题，这意味着等待水印可以正常前进，基于观察输入流中更大的时间戳。正如您所指出的，在您的用例中，这可能需要等待几个小时。

（2） 安排输入流包含保持活动状态的消息。这样，水印生成器将有证据（基于保持活动状态消息中的时间戳）它可以推进水印。您必须修改查询以忽略这些其他无关的事件。

(3)在到达作业已经完全摄取了所有的日常输入，但是还没有产生最终结果集的点时，停止作业并指定< code> - drain。这将通过管道发送一个值为MAX_WATERMARK的水印，这将关闭所有挂起的窗口。然后，您可以重新启动该作业。

（4）实现自定义水印策略，使用处理时间定时器检测空闲，并根据挂钟时间的流逝人为地推进水印。这将需要将表输入转换为 DataStream，在其中添加水印，然后转换回表以进行窗口化。有关这些转换的示例，请参阅 https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/。