I have a spark stream that reads every 5 seconds from Kafka. Each read cycle it creates a queue with InputSize = 10 records (for example). After a while I see a lot of batches with status "queued". I would prefer to process a large input size per batch. One way to achieve it is to increase the kafka read interval from 5 seconds to a larger duration. But I wanted to see if there is any way to dynamically combine multiple queued batches into a one large batch. For example: If my queues looks like this:
Batch Time Input Size Scheduling Delay (?) Processing Time (?) Output Ops: Succeeded/Total Status
2017/05/14 17:48:15 4 records - -
0/1 queued
2017/05/14 17:48:10 6 records - -
0/1 queued
2017/05/14 17:48:05 4 records - -
0/1 queued
Is there a way to pragmatically combine the above three batches into one single batch with input size = 14 records?
This is how I am creating the stream:
JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(
jssc,
String.class,
String.class,
StringDecoder.class,
StringDecoder.class,
kafkaParams,
topicsSet
);
Thank you for your help.