KStream 窗口聚合

Question

尝试合并多个 Kafka 流，聚合

val streams = requestStreams.merge(successStreams).merge(errorStreams)
                .groupByKey(Grouped.with(Serdes.String(), serdesConfig.notificationSerde()))
                .windowedBy(TimeWindows.of(Duration.ofMinutes(10)))
                .aggregate({ null }, StreamAggregators.notificationMetricAggregator, Materialized.`as`<String, NotificationMetric, WindowStore<Bytes, ByteArray>>("ag-store")
                        .withValueSerde(serdesConfig.notificationMetricSerde()))
                .toStream()

streams.to(notificationStreamsConfig.metricsTopic, Produced.with(WindowedSerdes.timeWindowedSerdeFrom(String::class.java, 10), serdesConfig.notificationMetricSerde()))

Answer 1

Kafka Streams默认使用连续更新处理模型。注意，聚合的结果是一个KTable。此结果表包含每个窗口的一行，每次处理新记录时，都会更新窗口（即表中的行）。

如果您调用Ktable#toStream（），您将获得表的更改日志流，其中包含表的每次更新的记录。

如果希望每个窗口只获得一个结果，可以使用< code>suppress()运算符来获得第二个< code>KTable，即< code>suppress()获取第一个< code>KTable的changelog流，并等待直到窗口关闭，然后只将最终结果插入其输出< code>KTable中。如果使用< code>suppress()，应该将上行窗口聚合的宽限期(默认值为24小时)设置为一个较低的值，即< code>TimeWindows.of(...).格蕾丝(...)。

有关更多详细信息，请查看此博客文章：https://www.confluent.io/blog/kafka-streams-take-on-watermarks-and-triggers