如何从kafka流中获得窗口聚合？

Question

我有一个事件流，我想聚集基于时间窗口。我的解决方案提供增量聚合，而不是在定时窗口上提供聚合。我读到过，这对于stream来说是正常的，因为它会以更改日志的形式给出结果。另外，在研究过程中，我遇到了两步窗口聚合与Kafka Streams DSL和如何发送最终的kafka-streams聚合结果的时间窗口Ktable？.但是第一篇文章中的解决方案有些过时（使用不推荐的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+receptence+regence+timed+windows wiki中给出的模式现在已经过时，因为它使用的是旧的API。（我使用的是kafka-streams 1.1.0版本）

Answer 1

问题是我的错误。上面的代码示例工作正常。但是在最后，我已经将ktable转换为kstream。这就是问题所在。转换为kstream也会导致输出中间结果。https://cwiki.apache.org/confluence/display/kafka/windowed+aggregations+over+receptence+regence+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");