如何从 kafka 流获取窗口聚合？

Question

我有一个事件流，我想根据时间窗口聚合这些事件。我的解决方案提供增量聚合，而不是在定时窗口上进行聚合。我已经读过这对于流来说是正常的，因为它会将结果作为更改日志。同样在研究期间，我遇到了 Kafka Streams DSL 的 2 步窗口聚合以及如何发送时间窗口 KTable 的最终 kafka-streams 聚合结果？. 但是第一篇文章中的解决方案有些过时（使用已弃用的 API）。我使用了那些已弃用的 API 中建议的新 API。这是我的解决方案

KStream<String, Event> eventKStream = summarizableData.mapValues(v -> v.getEvent());
    KGroupedStream<String, Event> kGroupedStream = eventKStream.groupBy((key, value) -> {
             String groupBy = getGroupBy(value, criteria);
             return groupBy;
    }, Serialized.with(Serdes.String(), eventSerde));


    long windowSizeMs = TimeUnit.SECONDS.toMillis(applicationProperties.getWindowSizeInSeconds());
    final TimeWindowedKStream<String, Event> groupedByKeyForWindow = kGroupedStream
            .windowedBy(TimeWindows.of(windowSizeMs)
                    .advanceBy(windowSizeMs));

但是，正如我之前所解释的，我的结果不是在特定时间窗口中给出的，而是作为增量聚合给出的。我需要我的数据按照 windowSize 中给出的指定时间输出。我也读到CACHE_MAX_BYTES_BUFFERING_CONFIG可以控制输出，但我需要一些可靠的解决方案适用于每种情况。另请注意，https: //cwiki.apache.org/confluence/display/KAFKA/Windowed+aggregations+over+successively+increasing+timed+windows wiki 中给出的模式现在已经过时，因为它使用旧的 API。（我使用的是 kafka-streams 1.1.0 版本）

Answer 1

问题是我的错误。上面的代码示例运行良好。但最后我已经将其转换KTable为KStream. 这就是问题所在。转换为KStream原因并输出中间结果。https://cwiki.apache.org/confluence/display/KAFKA/Windowed+aggregations+over+successively+increasing+timed+windows中给出的模式 工作正常。有问题的代码是，

// Aggregation

KTable<Windowed<String>, Event> results = groupedByKeyForWindow.aggregate(new AggregateInitiator(), new EventAggregator());

// This converstion causing changelog to output. Instead use next line.
KStream<String, AggregationMessage> aggregationMessageKStream = results.toStream((key, value) -> key.toString())
                .mapValues(this::convertToAggregationMessage).filter((k, v) -> v != null);

// output KTable to sample topic. But this output controlled by 
// COMMIT_INTERVAL_MS_CONFIG and CACHE_MAX_BYTES_BUFFERING_CONFIG parameters. 
// I'm using default values for these params.
results.to(windowedSerde, eventSerde,  "Sample");