为什么我看不到Kafka Streams reduce方法的任何输出？

Question

给出以下代码:

KStream<String, Custom> stream =  
    builder.stream(Serdes.String(), customSerde, "test_in");

stream
    .groupByKey(Serdes.String(), customSerde)
    .reduce(new CustomReducer(), "reduction_state")
    .print(Serdes.String(), customSerde);

我println在Reducer的apply方法中有一个声明,当我希望减少时,它会成功打印出来.但是,上面显示的最终打印语句不显示任何内容.同样,如果我使用to方法而不是print,我在目标主题中看不到任何消息.

在reduce语句之后我需要什么才能看到减少的结果？如果一个值被推送到输入,我不希望看到任何东西.如果按下具有相同键的第二个值,我希望减速器应用(它确实如此),并且我还期望减少的结果继续到处理管道中的下一步.如上所述,我在管道的后续步骤中没有看到任何内容,我不明白为什么.

Answer 1

从Kafka开始,0.10.1.0所有聚合运算符都使用内部重复数据删除缓存来减少结果KTable更改日志流的负载.例如,如果您使用相同的密钥直接计算和处理两个记录,则完整的更改日志流将是<key:1>, <key:2>.

使用新的缓存功能,缓存将接收<key:1>并存储它,但不会立即将其发送到下游.当<key:2>计算,它替换高速缓存中的第一项.根据缓存大小,不同密钥数,吞吐量和提交间隔,缓存会向下游发送条目.这发生在单个密钥条目的缓存逐出或缓存的完全刷新(向下游发送所有条目).因此,KTable更改日志可能只显示<key:2>(因为<key:1>重复数据删除).

您可以通过Streams配置参数控制缓存的大小StreamConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG.如果将值设置为零,则完全禁用缓存,KTable更改日志将包含所有更新(有效地提供预先0.10.1.0行为).

汇编文档包含更详细地解释缓存的部分:

• http://docs.confluent.io/current/streams/architecture.html#record-caches
• http://docs.confluent.io/current/streams/developer-guide.html#streams-developer-guide-memory-management