Использование KStreams для объединения записей

Question

Я работаю над своим первым Kafka Stream с использованием Java, и, по сути, я пытаюсь прочитать записи из одной темы, а затем объединить значения ряда этих записей в одну запись, которая будет записана в отдельную тему, т.е.серия записей в одну запись.В идеале эта партия будет контролироваться либо размером записи, либо количеством записей, но на этом этапе мы будем рады получить простой пример работы!Я думал, что функция KStream.groupBy будет основой для решения, но не может найти рабочий пример

Answer 1

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

KTable<String, List<Object>> aggregatedMetrics = eventStream
        .selectKey((k,v)-> k // Pick your key here)
        .groupByKey()
        .aggregate(() -> ArrayList::new, 
         (key, value, aggregate) -> aggregate.add(value), arrayListSerde());