kafka streams - how to set a new key for KTable

Question

I am new to Kafka Streams, I am using version 1.0.0. I would like to set a new key for a KTable from one of the values.

When using KStream, it cane be done by using method selectKey() like this.

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

However such method is missing in KTable. Only way is to convert given KTable to KStream. Any thoughts on this issue? Its changing a key against design of KTable?

Answer 1

If you want to set a new key, you need to re-group the KTable:

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

Because a key must be unique for a KTable (in contrast to a KStream) it's required to specify an aggregation function that aggregates all records with same (new) key into a single value.

Answer 2

@Matthias's answer led me down the right path, but I thought having a sample piece of code might help out here

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() documentation: https://kafka.apache.org/20/javadoc/org/apache/kafka/streams/kstream/KGroupedTable.html#aggregate-org.apache.kafka.streams.kstream.Initializer-org.apache.kafka.streams.kstream.Aggregator-org.apache.kafka.streams.kstream.Aggregator-org.apache.kafka.streams.kstream.Materialized-

Answer 3

For the ones who are using confluent 5.5.+ there is a method that allows extract the key from a stream and convert to a KTable directly:

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

Details can be found here

Answer 4

@Allen Underwood code helped me, had to make some changes if key is custom Pojo. As i was getting class cast exception. Below code worked

usersKTable.groupBy((k, v) -> KeyValue.pair(v.getCompositeKey(), v),Grouped.with(compositeKeySerde,valueSerde))
                .aggregate(
                        () -> null,
                        (applicationId, value, aggValue) -> value,
                        (applicationId, value, aggValue) -> value,
                        Materialized.with(compositeKeySerde, valueSerde)
                );

Answer 5

I don't think the way @Matthias described it is accurate/detailed enough. It is correct, but the root cause of such limitation(exists for ksqlDB CREATE TABLE syntax as well) is beyond just sheer fact that the keys must be unique for KTable.

The uniqueness in itself doesn't limit KTables. After all, any underlying topic can, and often does, contain messages with the same keys. KTable has no problem with that. It will just enforce the latest state for each key. There are multiple consequences of this, including the fact that KTable built from aggregated function can produce several messages into its output topic based on a single input message...But let's get back to your question.

So, the KTable needs to know which message for a specific key is the last message, meaning it's the latest state for the key.

What ordering guarantees does Kafka have? Correct, on per partition basis.

What happens when messages are re-keyed? Correct, they will be spread across partitions very different from the input message.

So, the initial messages with the same key were correctly stored by the broker itself into the same partition(if you didn't do anything fancy/stupid with your custom Partitioner) That way KTable can always infer the latest state.

But what happens if the messages are re-keyed inside Kafka Streams application in-flight?

They will spread across partitions again, but with a different key now, and if your application is scaled out and you have several tasks working in parallel you simply can't guarantee that the last message by a new key is actually the last message as it was stored in the original topic. Separate tasks don't have any coordination like that. And they can't. It won't be efficient otherwise.

As a result, KTable will lose its main semantic if such re-keying were allowed.