java - kafka 流 - 如何为 KTable 设置新 key

Question

我是 Kafka Streams 的新手，我使用的是 1.0.0 版。我想从其中一个值中为 KTable 设置一个新键。

使用 KStream 时，可以通过像这样使用 selectKey() 方法来完成。

kstream.selectKey ((k,v) -> v.newKey)

但是，KTable 中缺少这种方法。唯一的方法是将给定的 KTable 转换为 KStream。关于这个问题的任何想法？它改变了KTable的设计的一个关键？

Answer 1

如果要设置新的key，需要重新对KTable进行分组:

KTable newTable = table.groupBy(/*put select key function here*/)
                       .aggregate(...);

因为 KTable 的键必须是唯一的(与 KStream 相比)，所以需要指定一个聚合函数，将具有相同(新)键的所有记录聚合为单个值。