使用 Kafka Stream 和 Windowed Stream 生成 Statistic 事件，但如果有最近的事件，则跳过一些窗口事件答案

【问题标题】：Produce Statistic event with Kafka Stream and Windowed Stream, but skip some windowed if there is some more recent one使用 Kafka Stream 和 Windowed Stream 生成 Statistic 事件，但如果有最近的事件，则跳过一些窗口事件
【发布时间】：2021-03-01 02:24:34
【问题描述】：

对于过去 24 小时的事件，我想生成一个统计事件以在某处实时显示。所以，目前的行为是这样的：

每分钟，我都会聚合最近 24 小时的事件以添加列表对象
我临时计算完整列表并生成最终统计对象
我将这个最终统计对象推送到新主题

我正在使用 Kafka Stream 和 Spring Boot。

它运作良好，我有很好的计算能力，并且该事件在开发中很好地产生。问题是我在生产时，源事件主题包含太多数据。

如果我的应用程序停止一天或几分钟。当应用程序重新启动时，我的应用程序尝试恢复历史记录。 Kafka 流从最后一个偏移量继续处理，并且需要花费大量时间来赶上他的延迟。事实上，我并不关心历史。我不需要昨天或过去 24 小时减去 1 小时的统计对象，我只想从现在重新计算到过去 24 小时，就是这样。

如果应用程序运行正常，则相同，但处理统计事件有一些延迟。滞后增加和增加。如果延迟变得太重要，我会 Kafka Stream 自动跳过时间窗口并只计算最后一个。

你认为 Kafka Stream 可以做到吗？提前致谢。

    /**
 * Every minute, we collect all events on the last day and we publish a new statistic event.
 * 
 * @param streamsBuilder
 * @return
 */
@Bean
public KStream<String, MySourceEvent> kstreamMySourceEventStatistique(final StreamsBuilder streamsBuilder) {

    // We create the stream to consume machine-state topic.
    KStream<String, MySourceEvent> kstreamStat = streamsBuilder
            .<String, MySourceEvent>stream("my-source-topic", Consumed
                    .with(Serdes.String(), KafkaUtils
                            .jsonSerdeForClass(MySourceEvent.class)));

    // For this stream, every minute, we take all events in the last 24h, and we aggregate them into TemporaryStatistiqueEvent
    KTable<Windowed<String>, TemporaryStatistiqueEvent> aggregatedStream = kstreamStat 
            .groupByKey(Grouped
                    .with(Serdes.String(), KafkaUtils
                            .jsonSerdeForClass(MySourceEvent.class)))
            .windowedBy(TimeWindows
                    .of(Duration.ofDays(1))
                    .advanceBy(Duration.ofMinutes(1))
                    .grace(Duration.ofSeconds(0)))
            .<TemporaryStatistiqueEvent>aggregate(() -> new TemporaryStatistiqueEvent(), (key, value, logAgg) -> {
                logAgg.add(value); //I add the event in my TemporaryStatistiqueEvent object
                return logAgg;
            }, Materialized
                    .<String, TemporaryStatistiqueEvent, WindowStore<Bytes, byte[]>>as("temporary-stats-store")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(KafkaUtils.jsonSerdeForClass(TemporaryStatistiqueEvent.class))
                    .withRetention(Duration.ofDays(1)))
            .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()));
    // Now, we gave an aggregate on last 24h, we compute the statistic and push FinalStatisticEvent object in a new topic
    aggregatedStream
            .toStream()
            .map(new KeyValueMapper<Windowed<String>, TemporaryStatistiqueEvent, KeyValue<String, PlcStatMachineState>>() {
                @Override
                public KeyValue<String, FinalStatisticEvent> apply(final Windowed<String> key, final TemporaryStatistiqueEvent temporaryStatistiqueEvent) {
                    ZonedDateTime zdt = ZonedDateTime.ofInstant(Instant.ofEpochMilli(key.window().end()), ZoneOffset.UTC);
                    return new KeyValue<>(key.key(), temporaryStatistiqueEvent.computeFinalStatisticEvent(zdt));
                }
            })
            .to("final-stat-topic", Produced.with(Serdes.String(), KafkaUtils.jsonSerdeForClass(FinalStatisticEvent .class)));

    return kstreamStat;
}

【问题讨论】：

标签： apache-kafka apache-kafka-streams spring-kafka

【解决方案1】：

这是一个棘手的问题...

对于离线和重新启动的情况，您可以尝试在重新启动应用程序之前操纵开始偏移量（即提交的偏移量）。使用bin/kafka-consumer-group.sh，您可以“按时间搜索”，从而“向前跳到 now-minus-24h”。

对于应用滞后的情况，情况更加复杂。也许您可以在您的程序中使用“动态过滤器”（访问记录元数据，如其时间戳，您可以使用flatTransformValues 来实现过滤器）作为程序的第一步，它会删除太旧的记录？

【讨论】：