Ограничить размер пакета kafka при использовании Spark Structured Streaming

Question

У нас есть некоторые исторические данные, поставленные в очередь по нашим темам, мы не хотим обрабатывать все эти данные в одном пакете, так как это сложнее сделать (и, если это не удастся, оно должно начаться снова!).

Кроме того, знание того, как контролировать размер пакета, было бы весьма полезно при настройке заданий.

При использовании DStreams способ контролировать размер пакета как можно точнее - Limit Kafkaразмер пакетов при использовании Spark Streaming

Тот же подход, т. е. установка maxRatePerPartition, а затем настройка batchDuration, чрезвычайно громоздка, но работает с DStream, но не работает со структурированной потоковой передачей.

В идеале я хотел бы знать о конфигурации, такой как maxBatchSize и minBatchSize, где я могу просто установить количество записей, которое я хочу.

Answer 1

Эта опция конфигурации maxOffsetsPerTrigger:

Ограничение скорости на максимальное количество смещений, обрабатываемых за интервал запуска.Указанное общее количество смещений будет пропорционально разделено на topicPartitions различного объема.

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