Kafka流聚合器-如何设置时间等待在聚合之前吐出消息？

Question

(0_10,..)，(0_11,..)-->(0,[10])(0,[10,11])

我想知道如何控制聚合时间窗口，这样它就不会为每个传入的消息吐出一条消息，而是等待并聚合其中的一些消息。想象一下流应用程序使用这些消息：

• (0_10,..）
• (1_11,..)
• (0_13,..）

null

我的代码很简单

builder
    .table(keySerde, valueSerde, sourceTopic)
    .groupBy(StreamBuilder::groupByMapper)
    .aggregate(
        StreamBuilder::aggregateInitializer,
        StreamBuilder::aggregateAdder,
        StreamBuilder::aggregateSubtractor)
    .to(...);

目前，它有时会批量聚合，但不确定如何调整它：

{"Aggregate":[100]}
{"Aggregate":[100,300,301,302]}
{"Aggregate":[100,300,301,302,404]}

Answer 1

我想知道如何控制聚合时间窗口，这样它就不会为每个传入的消息吐出一条消息，而是等待并聚合其中的一些消息。

这是不可能的Kafka流的窗口。一般来说，Kafka Streams窗口不会“关闭”或“结束”，因为一旦窗口“关闭”，你就不能告诉它产生最终结果（没有这样的概念）。这是为了适应迟来的结果。当消息到达聚合窗口时，您将看到更新。Kafka流输出更新的频率取决于缓存（见下文）。有关更多信息，请参见：如何发送时间窗口Ktable的最终kafka-streams聚合结果？

目前，它有时会批量聚合，但不确定如何调整它：