как обрабатывать данные в блоках / пакетах с потоками кафки?

Question

Для многих ситуаций в больших данных предпочтительно работать с небольшим буфером записей на ходу, а не с одной записью за раз.

Естественным примером является вызов некоторого внешнего API, который поддерживает пакетную обработку дляэффективность.

Как мы можем сделать это в Кафка-Стримс?В API я не могу найти ничего похожего на то, что я хочу.

Пока у меня есть:

builder.stream[String, String]("my-input-topic")
.mapValues(externalApiCall).to("my-output-topic")

Что я хочу:

builder.stream[String, String]("my-input-topic")
.batched(chunkSize = 2000).map(externalBatchedApiCall).to("my-output-topic")

В Scalaи Akka Streams функция называется grouped или batch.В Spark Structured Streaming мы можем сделать mapPartitions.map(_.grouped(2000).map(externalBatchedApiCall)).

Answer 1

Кажется, еще не существует. Смотреть это пространство https://issues.apache.org/jira/browse/KAFKA-7432

Answer 2

Подозреваю, если поток Kafka поддерживает окна фиксированного размера, как и другие инструменты в данный момент.
Но существуют окна на основе времени, поддерживаемые потоками kafka.https://kafka.apache.org/11/documentation/streams/developer-guide/dsl-api.html#windowing

Вместо количества записей вы можете определить размер окна со временем.

1. Падающие временные окна
2. Скользящее временное окно
3. Окно сеанса
4. Прыгающее временное окно

В вашем случае, Tumbling Time Window можно использовать.Это не перекрывающиеся временные окна фиксированного размера.

Например, падающие окна размером 5000 мс имеют предсказуемые границы окон [0; 5000), [5000; 10000), ... -а не [1000; 6000), [6000; 11000), ... или даже что-то «случайное», подобное [1452; 6452), [6452; 11452), ....