Kafka 流中的计数操作后出现意外键。

Question

我在 kafka 流中有以下逻辑流程：

stream.map((nullKey, rawData) -> KeyValue.pair(somekey(rawData), rawData))
            .filter((k, v) -> somefilterning(v))
            .groupByKey()
            .count(TimeWindows.of(3600).until(TimeUnit.SECONDS.toMillis(7200)), "someStateStore")
            .foreach((k, v) -> print(k.window().start()));

传递给 count 方法的参数与上面显示的相同，因此，它应该按（逻辑）每小时 window 聚合键，即所有键都在 11:00 和 12:00应该聚合在一起。

为了验证这一点，我在 foreach 调用中打印了 window 开始时间（或者至少这是我期望的）。我的 COMMIT_INTERVAL_MS_CONFIG 设置为 2 分钟，因此计数输出应每 2 分钟刷新一次，我希望 key.window.start() 在刷新之间保持不变（当然假设刷新的逻辑时间相同）。

相反，我看到许多不同的 start() 值：

     timestamp    time
0   1508068706  11:58:26
1   1508068713  11:58:33
2   1508068720  11:58:40
3   1508068728  11:58:48
4   1508068735  11:58:55
5   1508068742  11:59:02

这与逻辑小时 (11:00) 有很大不同。它也与 2 分钟的提交间隔无关，因为您可以看到时间戳有 8 秒的差异。

Answer 1

这个答案有多个部分：

首先，Kafka Streams 会连续更新一个 window 并在每次更新后发出当前结果（也比较 and https://www.confluent.io/blog/watermarks-tables-event-time-dataflow-model/）
其次，内部使用缓存（cf.https://docs.confluent.io/current/streams/developer-guide.html#memory-management）对同一个键的连续更新进行去重，每次需要为新记录制作space时发出数据（基于LRU战略）;因此，它可以在提交之间发出——它也会在提交时完全刷新
第三，您指定的 window 大小为 3600 并且此值以毫秒为单位 -- 而不是秒；我猜你想做 TimeWindows.of(TimeUnit.SECONDS.toMillis(3600))

Kafka 流中的计数操作后出现意外键。

Unexpected key after count operation in Kafka streams.

apache-kafka-streams