[SPARK-18371][Streaming] Spark Streaming backpressure generates batch with large number of records

[arzt]

What changes were proposed in this pull request?

Omit rounding of backpressure rate. Effects:

• no batch with large number of records is created when rate from PID estimator is one
• the number of records per batch and partition is more fine-grained improving backpressure accuracy

How was this patch tested?

This was tested by running:

• mvn test -pl external/kafka-0-8
• mvn test -pl external/kafka-0-10
• a streaming application which was suffering from the issue

@JasonMWhite

The contribution is my original work and I license the work to the project under the project’s open source license

[JasonMWhite]

Code looks sound. Could you add or modify a test to illustrate/verify?

[srowen]

CC @koeninger
The thing is, it seems like rates are intentionally not floating point here, but I don't know the history of it.

[srowen]

(Given your change, I think double becomes more reasonable than float.

[koeninger]

How do you read 0.1 of a kafka message for a given partition of a given batch?

Ultimately the floor for a rate limit, assuming one is set, needs to be 1 message per partition per batch, not a fraction, which is why it's a long.

If you want to delay that conversion by keeping it as a double as long as possible, that makes sense, but the lines like

(secsPerBatch * limit).toLong

probably need attention too.

[arzt]

Thanks for your valuable feedback. I added tests as suggested by @JasonMWhite and used toDouble. @koeninger the estimated rate is per second summed over all partitions, is it? The batch time usually is longer. So even values less than 1 but greater than 0 for backpressureRate can make sense for one partition. The casting to long is only needed when the absolute number of messages is computed, but even this number can be zero for some partitions, e.g. when there is no lag. I hope I am not confused here. There is also a test covering this.

[arzt]

To run tests or debug using IntelliJ:
mvn test -DforkMode=never -pl external/kafka-0-8 "-Dsuites=org.apache.spark.streaming.kafka.DirectKafkaStreamSuite maxMessagesPerPartition"

[koeninger]

@arzt It's entirely possible to have batch times less than a second, and I'm not sure I agree that the absolute number of messages allowable for a partition should ever be zero.

So to put this another way, right now effectiveRateLimitPerPartition is a Map[TopicPartition, Long], which matches the return value of the function maxMessagesPerPartition.

You're wanting to change effectiveRateLimitPerPartition to a Map[TopicPartition, Double], which is probably a good idea, and should fix the bug around treating a very small rate limit as no limit.

But it still needs to be converted to Map[TopicPartition, Long] before returning. Calling .toLong is probably not the right thing to do there, because 0.99 will get truncated to 0.

I think one message per partition per batch is the minimum reasonable rate limit, otherwise particular partitions may not make progress. The relative lag calculation might take care of that in future batches, but it still seems questionable, even if it's a corner case.

[arzt]

@koeninger I agree that assuming a long batch size is wrong, not sure whether it even matters.
But what if for one partition there is no lack in the current batch? Then fetching 1 message for this partition from kafka, is you suggest, would fail. So here zero makes sense in my eyes. This is also the old behaviour if rate > 1 and lag == 0 here.
Further, I think that truncating 0.99 to 0 messages per partition is also the right thing to do, as one cannot be sure that there is one message available if (secsPerBatch * limit) < 1.0. And as you say, in a future batch it is very like to become greater than 1.0.
Do you agree?

[koeninger]

Have you read the function def clamp? Rate limit of 1 should not imply an attempt to grab 1 message even if it doesn't exist.

…

On Apr 27, 2017 11:01, "Sebastian Arzt" ***@***.***> wrote: @koeninger <https://github.com/koeninger> I agree that assuming a long batch size is wrong, not sure whether it even matters. But what if for one partition there is no lack in the current batch? Then fetching 1 message for this partition from kafka, is you suggest, would fail. So here zero makes sense in my eyes. This is also the old behaviour if rate > 1 and lag == 0 here <https://github.com/apache/spark/blob/master/external/kafka-0-8/src/main/scala/org/apache/spark/streaming/kafka/DirectKafkaInputDStream.scala#L107> . Further, I think that truncating 0.99 to 0 messages per partition is also the right thing to do, as one cannot be sure that there is one message available if (secsPerBatch * limit) < 1.0. And as you say, in a future batch it is very like to become greater than 1.0. Do you agree? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#17774 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAGAB1X8NCNqECUlx9X54DSAmbnHmHdAks5r0LvggaJpZM4NJAVA> .

[JasonMWhite]

I think @koeninger's suggestion is valid. effectiveRateLimitPerPartition is the upper bound on the number of messages per partition per second, and maxMessagesPerPartition sets an upper bound on the number of messages to be retrieved per partition per batch window.

Making effectiveRateLimitPerPartition a float will allow it to handle properly rates of < 1/partition/s, so this is definitely a good idea. maxMessagesPerPartition must still be an integer, as you can't retrieve partial messages. All agreed there.

Setting maxMessagesPerPartition to have a minimum of 1 message per partition per batch window is a good safe value to allow progress in all cases. If there isn't 1 message to retrieve, clamp will prevent it from attempting to retrieve an invalid message.

[arzt]

I changed the max messages per partition to be at least 1. Agreed?

[JasonMWhite]

The actual result should be deterministic, why not check the correct value instead of just not None ?

[arzt]

After omitting the case of zero messages per topic, one of the tests is redundant. I removed it from each DirectKafkaStreamSuite.

[JasonMWhite]

Tests have some fairly repetitive code, but not sure if that's a problem or not. Looks good to me.

[koeninger]

LGTM pending jason's comments on tests

[felixcheung]

Jenkins, ok to test

[SparkQA]

Test build #76256 has finished for PR 17774 at commit d4a7867.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #76263 has finished for PR 17774 at commit c98b9a4.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[arzt]

Sorry for being inactive. All good with this?

[JasonMWhite]

LGTM

[arzt]

@felixcheung will this be merged?

[felixcheung]

@brkyvz @zsxwing

[arzt]

It's been a while. What can I do to draw some attention to this request? Is this issue not relevant enough? Thanks for reconsideration @felixcheung @brkyvz @zsxwing

[SparkQA]

Test build #78618 has finished for PR 17774 at commit 16b9aaf.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[felixcheung]

I think @tdas @zsxwing should comment if this is the right direction...

[pptaszynski]

I am looking forward this one to be merged. we are suffering from the issue it resolved quite badly. It effectively makes the back-pressure not working for us at all.

[koeninger]

LGTM
@tdas @zsxwing absent any objections from you in the next couple of days, I'll merge this

[felixcheung]

Jenkins, ok to test

[felixcheung]

pending tests

[SparkQA]

Test build #88226 has finished for PR 17774 at commit 1acbe4c.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[koeninger]

merged to master
Thanks @arzt !