Kafka Streams - агрегирование и объединение пользователей с адресами

Я спрашиваю себя, удалит ли KGroupedStream некоторые потоки через некоторое время или нет.

Он ничего не удалит.

Если я понимаю остальную часть вашего вопроса, вы спрашиваете, как работает оператор aggregate(). Он использует локальное хранилище состояний (реализованное с помощью RocksDB) для хранения <userId, X>, где X - это то, что возвращает ваш агрегированный UDF ((user, address, queue) -> { }), т. Е. Он должен быть X == queue). Таким образом, каждая входная запись выполняет локальный поиск в RocksDB для получения текущего queue, обновляет его, записывает обратно в RocksDB и отправляет его вниз по потоку в ваш оператор to(), который записывает его также в тему результатов.

Также прочтите документацию для получения дополнительных сведений: https://kafka.apache.org/21/documentation/streams/ Также есть много других материалов о Kafka Streams и о том, как они работают в Интернете (сообщения в блогах, записи выступлений, слайды ...)

Интересно, что результат после соединения (в сжатой теме с именем user_addresses) содержит больше записей, чем таблица записей пользователя. Я посмотрел глубже и увидел, что у пользователя с одним и тем же ключом есть несколько вхождений (несколько смещений). При наименьшем смещении у этого пользователя нет адресов, затем при более высоком смещении он имеет один адрес в своем списке, а при наибольшем смещении он имеет два адреса в своем списке. Я снова спрашиваю себя, почему старые смещения не удаляются автоматически, когда я использую уплотненную тему. Работает ли уплотнение Кафки как сборщик мусора, который впоследствии удаляет данные? Что делать, если я ищу ключ, получу ли я ключ с наибольшим смещением?

Сжатие выполняется асинхронно в фоновом режиме, но не сразу. Также обратите внимание, что тематические (или, точнее,) разделы разбиты на «сегменты», а активный сегмент никогда не сжимается (размер сегмента по умолчанию - 1 ГБ). Вы можете настроить размер сегмента и то, как часто запускается сжатие (дополнительные сведения см. В документации: https://kafka.apache.org/documentation/#compaction).

Что делать, если я ищу ключ, получу ли я ключ с наибольшим смещением?

Не уверен, что вы имеете в виду. Kafka допускает только последовательное чтение, но не поиск ключей. Таким образом, вам нужно будет прочитать тему от начала до конца, чтобы найти последнюю версию ключа. Если вы обратитесь к функции «Интерактивные запросы» Kafka Streams, она будет запрашивать локальную базу данных RocksDB и, таким образом, содержать последнюю запись для каждого ключа.

Мой вопрос в том, подходит ли мой подход для достижения этой цели.

Да, с одной важной деталью, связанной с

В чем внутренняя разница между KGroupedStream и KGroupedTable?

Поскольку вы вводите тему - это сжатая тема, в которой используются ключи (userId,addressId), вы должны читать ее как table() (а не stream()):

KTable<GenericRecord, ArrayList<GenericRecord>> aggregatedAddresses =
    builder.table("address-topic")
      .selectKey(...) //Selecting USERID as key - this generates KStream
      .groupBy(...) //Select USERID as and group by USERID
      .aggregate(...) //Aggregating by USERID as key - this generates KTable
      .to("aggregated_addresses"); //KTable with USERID as key 

Разница в том, что если вы читаете тему KStreams, это интерпретируется как «факты», и поэтому семантика удаления отсутствует. Однако вводимая вами тема содержит записи «обновлений» и, следовательно, должна быть потребителем как таковая. KGroupedStream и KGroupedTable - это просто промежуточные объекты в API, которые также подразумевают семантику «факт» и «обновление». Опять же, проверьте документы и другие материалы в Интернете для получения более подробной информации.