如何让Kafka Streams每1小时窗口每键发送一条记录？

Question

我正在编写一个Kafka Streams应用程序。它执行以下步骤“1）消耗输入数据2）在1小时窗口内根据新密钥消重记录3）重新选择密钥4）在1小时窗口内计数密钥5）发送到下游。

我是Kafka流的新手。我的理解是，为了将窗口保持为1小时，我将 commit.interval.ms 也设置为1小时。这是正确的做法吗？

一旦我部署了我的应用程序与真实的流量，它似乎应用程序不断发送消息，而我认为它每小时只会发送一堆消息？

感谢任何帮助！！

我的配置:

commit.interval.ms = 3600000
request.timeout.ms = 600000
retries = 20
retry.backoff.ms = 1000    
cache.max.bytes.buffering = 10485760

// dedupe by new key per window(1hr)
 stream = inputStream
        .selectKey(... )
        .groupByKey()
        .windowedBy(TimeWindows.of(TimeUnit.MINUTES.toMillis(60)))
        // only keep the latest event for each customized key
        .reduce((event1, event2) -> event2)
        .toStream()
        .groupByKey()
        .windowedBy(TimeWindows.of(TimeUnit.MINUTES.toMillis(60)))
        .reduce((event1, event2) -> {
            long count1 = event1.getCount();
            long count2 = event2.getCount();
            event2.setCount(count1 + count2);
            return event2;
        })
        .toStream()
        .to(OUTPUT_TOPIC);

Answer 1

1. 我建议您使用最新版本的kafka提供的精确一次保证。使用它，您不必担心重复消息的删除。https://www.baeldung.com/kafka-exactly-once
2. 配置生产者配置：特别是buffer.memory

Answer 2

我是Kafka流的新手。我的理解是，为了将窗口保持为1小时，我将 commit.interval.ms 也设置为1小时。这是正确的做法吗？

提交间隔与您的处理逻辑无关。

您可能希望研究一下< code>suppress()运算符。下面的block post也可能有所帮助:

• https://www.confluent.io/blog/watermarks-tables-event-time-dataflow-model/
• https://www.confluent.io/blog/kafka-streams-take-on-watermarks-and-triggers

Kafka Streams的处理模型是连续的，默认情况下会发送连续的结果更新。这就是为什么基本上每个输入消息都会得到一条输出消息，因为处理输入消息会修改结果。