Потоки kafka - как установить новый ключ для KTable

Если вы хотите установить новый ключ, вам нужно перегруппировать KTable:

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

Поскольку ключ должен быть уникальным для KTable (в отличие от KStream), необходимо указать функцию агрегирования, которая объединяет все записи с одним и тем же (новым) ключом в одно значение.

@ Ответ Матиаса привел меня по правильному пути, но я подумал, что здесь может помочь образец кода

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-org.apache.kafka.streams.kstream.Initializer-org.apache.kafka.streams. kstream.Aggregator-org.apache.kafka.streams.kstream.Aggregator-org.apache.kafka.streams.kstream.Materialized-

Код @Allen Underwood мне помог, пришлось внести некоторые изменения, если ключом является пользовательский Pojo. Поскольку я получал исключение приведения класса. Ниже код работал

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)
                );

Для тех, кто использует confluent 5.5. +, Есть метод, позволяющий извлечь ключ из потока и напрямую преобразовать его в KTable:

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

Подробности можно узнать здесь

Я не думаю, что описание @Matthias достаточно точным / подробным. Это правильно, но основная причина такого ограничения (существует и для синтаксиса ksqlDBCREATE TABLE) выходит за рамки простого факта, что ключи должны быть уникальными для KTable.

Уникальность сама по себе не ограничивает KTables. В конце концов, любая основная тема может содержать и часто содержит сообщения с одинаковыми ключами. У KTable с этим проблем нет. Он просто применяет последнее состояние для каждого ключа. Это имеет множество последствий, включая тот факт, что KTable, созданный из агрегированной функции, может создавать несколько сообщений в своей выходной теме на основе одного входного сообщения ... Но давайте вернемся к вашему вопросу.

Итак, KTable необходимо знать, какое сообщение для определенного ключа является последним сообщением, то есть это последнее состояние ключа.

Какие гарантии заказа есть у Kafka? Правильно, по разделам.

Что происходит, когда сообщения изменяются? Верно, они будут распределены по разделам, очень отличным от входного сообщения.

Итак, исходные сообщения с одним и тем же ключом были правильно сохранены самим брокером в том же разделе (если вы не сделали ничего необычного / глупого с вашим пользовательским Partitioner) Таким образом, KTable всегда может определить последнее состояние.

Но что произойдет, если сообщения перепрограммированы внутри приложения Kafka Streams на лету?

Они снова будут распространяться по разделам, но теперь с другим ключом, и если ваше приложение масштабировано и у вас есть несколько задач, работающих параллельно, вы просто не можете гарантировать, что последнее сообщение по новому ключу на самом деле является последним сообщением, поскольку оно было сохранено в исходной теме. У отдельных задач нет такой координации. А они не могут. В противном случае это не будет эффективно.

В результате KTable потеряет свою основную семантику, если такое изменение ключей было разрешено.