Kafka Streams 与时间窗口聚合的问题

Posted 2023-02-26

技术标签:

【中文标题】Kafka Streams 与时间窗口聚合的问题【英文标题】：Kafka KStreams Issue in Aggregation with Time Window 【发布时间】：2019-10-15 07:18:18 【问题描述】：

我对 KStreams 聚合和窗口有疑问。我想将一条记录聚合到一个具有相同键的记录列表中，只要它位于时间窗口内。我选择 SessionWindows 是因为我必须在会话中使用移动窗口：假设记录 A 在 10:00:00 到达；然后具有相同键的所有其他记录到达在 10 秒窗口时间内（直到 10:00:10）将进入同一会话，请记住，如果它在 10:00:03 到达，窗口将移动到 10:00:13（+10 秒）。

这导致我们从给定键收到的最后一条记录开始有一个 +10 秒的移动窗口。

现在的问题是：我想获得最后的聚合结果。我使用 .suppress() 表示我不想要任何中间结果，我只想要窗口关闭时的最后一个结果。这工作不正常，因为虽然它没有发送任何中间聚合结果，但当时间窗口结束时，我没有得到任何结果。我注意到，为了接收它，我需要发布另一个消息进入主题，这在我的情况下是不可能的。

阅读 .suppress() 我得出的结论是，它可能不是实现我想要的方式，这就是为什么我的问题是：如何强制关闭窗口并发送最新的聚合计算结果？

@StreamListener(ExtractContractBinding.RECEIVE_PAGE)
@SendTo(ExtractCommunicationBinding.AGGREGATED_PAGES) 
public KStream<String, List<Records>> aggregatePages(KStream<?, Record> input)  
    input.map(this::getRecord)
            .groupBy(keyOfElement)
            .windowedBy(SessionWindows.with(Duration.ofSeconds(10L)).grace(Duration.ofSeconds(10L)))
            .aggregate(...do stuff...)
            .suppress(Suppressed.untilWindowCloses(unbounded()))
            .toStream()
            .map(this::createAggregatedResult);

【问题讨论】：

【参考方案1】：

简而言之，发生这种情况的原因是因为在 KStreams 和大多数其他计算聚合的流处理引擎中，时间基于事件时间工作。

https://kafka.apache.org/0101/documentation/streams#streams_time

换句话说，在新消息到达超出您的时间窗口 + 宽限时间（导致迟到的消息）之前，该窗口无法关闭。

此外，根据我最近编写的一些单元测试，我倾向于认为第二条消息需要与前一条消息位于同一分区中，以便事件时间向前推进。在实践中，当您在生产环境中运行并且大概每秒处理数百条消息时，这会变得不明显。

我还要补充一点，您可以实现自定义时间戳提取器，它允许您根据特定消息所在的时间窗口进行细粒度控制。