Kafka流-如何为KTable设置新密钥

Question

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

在使用KStream时，可以使用如下方法selectKey（）完成。

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

然而，KTable中缺少这种方法。唯一的方法是将给定的KTable转换为KStream。对这个问题有什么想法吗？它改变了KTable设计的关键？

Answer 1

有一种方法可以从流中提取密钥并直接转换为KTable：

       KTable<String, User> userTable = builder
            .stream("topic_name", Consumed.with(userIdSerde, userSerde))
            .selectKey((key, value) -> key.getUserId())             
            .toTable( Materialized.with(stringIdSerde, userSerde));

详情可以在这里找到

Answer 2

@马蒂亚斯的回答让我走上了正确的道路，但我认为有一段代码样本可能会有所帮助

final KTable<String, User> usersKeyedByApplicationIDKTable = usersKTable.groupBy(
        // First, going to set the new key to the user's application id
        (userId, user) -> KeyValue.pair(user.getApplicationID().toString(), user)
).aggregate(
        // Initiate the aggregate value
        () -> null,
        // adder (doing nothing, just passing the user through as the value)
        (applicationId, user, aggValue) -> user,
        // subtractor (doing nothing, just passing the user through as the value)
        (applicationId, user, aggValue) -> user
);

KGroupedTable aggregate（）文档：https://kafka.apache.org/20/javadoc/org/apache/kafka/streams/kstream/KGroupedTable.html#aggregate-组织。阿帕奇。Kafka。溪流。kstream。初始化器组织。阿帕奇。Kafka。溪流。kstream。聚合器组织。阿帕奇。Kafka。溪流。kstream。聚合器组织。阿帕奇。Kafka。溪流。kstream。具体化-

Answer 3

如果要设置新密钥，需要对KTable进行重新分组：

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

因为一个键对于一个KTable（与KStream相反）必须是唯一的，所以需要指定一个聚合函数，将具有相同（新）键的所有记录聚合为一个值。

自Kafka2.5以来，Kafka流还支持KStream#toTable（）操作符。因此，也可以执行表格。toStream（）。选择键（…）。toTable（）。这两种方法都有优点和缺点。

使用toTable（）的主要缺点是，它会根据新键重新分区输入数据，这会导致交错写入重新分区主题，从而导致无序数据。虽然第一种方法通过group By（）使用相同的实现，但使用聚合函数有助于解决“冲突”的明确性。如果使用toTable（）运算符，则会完成基于重新分区主题偏移顺序的“盲”上插（这实际上类似于其他答案中的代码示例）。

例子：

Key | Value
 A  | (a,1)
 B  | (a,2)

如果您在a上重新输入，您的输出表将是两个表中的一个（但它没有定义为一个）：

Key | Value          Key | Value
 a  | 1               a  |  2

“重新设置”表的操作在语义上总是定义不清的。