Is there a way to combine spark's multiple waiting queues into one?

Viewed 173

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.

0 Answers
Related