This repository was archived by the owner on May 9, 2024. It is now read-only.
[SPARK-8834] Add backpressure-based dynamic throttling to Spark Streaming #14
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
huitseeker
changed the title
- RateEstimator introduces API for computing max rate of message ingestion from batch updates - RateController sends batch completion updates to an estimator, and maintains a max speed estimate - the RateController test is in ignore because of long running time
huitseeker
force-pushed
the
DirectBackPressure2
branch
from
fdb5cf2 to
33e3098
Compare
| @@ -77,6 +77,10 @@ private[streaming] class ReceiverSupervisorImpl( | |||
| case CleanupOldBlocks(threshTime) => | |||
| logDebug("Received delete old batch signal") | |||
| cleanupOldBlocks(threshTime) | |||
| case RateLimitUpdate(eps) => { | |||
| blockGenerator.updateRate(eps.toInt) | |||
| logDebug(s"Received update for $streamId : $elemsPerBlock") | |||
There was a problem hiding this comment.
elemsPerBlock is not defined. I guess it should be eps here.
There was a problem hiding this comment.
Crap, and I thought I was helping by adding one logging commit :-/
|
Code looks |
| if (ratePerSec > 0) { | ||
| protected def maxMessagesPerPartition: Option[Long] = { | ||
| val effectiveRatePerSec = | ||
| rateController.map(_.getLatestRate()).filter(_ > 0).fold(ratePerSec) {_ min _} |
There was a problem hiding this comment.
I get a compilation error on this line, but I'm not sure why.
[error] /home/ubuntu/dev/spark/spark-ops/release_spark/target/spark/external/kafka/src/main/scala/org/apache/spark/streaming/kafka/DirectKafkaInputDStream.scala:80: wrong number of parameters; expected = 1
[error] rateController.map(_.getLatestRate()).filter(_ > 0).fold(ratePerSec) {_ min _}
[error] ^
[error] one error found
[error] Compile failed at Jul 10, 2015 2:55:16 PM [1.198s]
There was a problem hiding this comment.
Couldn't reproduce, but I think I know why, wait a sec.
huitseeker
force-pushed
the
DirectBackPressure2
branch
4 times, most recently
from
f533773 to
c21d7c4
Compare
|
Refer to this link for build results (access rights to CI server needed): Build Log Test FAILed. |
This RateController maintains a prescriptive max speed estimate from a (possibly custom) rate estimator. Dynamically rate-limits this DStream if it is a ReceiverInputDStream
As with the RateLimiter, the global configuration parameter is taken as an absolute limit.
|
retest this please |
huitseeker
force-pushed
the
DirectBackPressure2
branch
from
c21d7c4 to
a62a31e
Compare
|
Refer to this link for build results (access rights to CI server needed): Build Log Test FAILed. |
|
completely unrelated flaky test in spark SQL's logical plans failing. |
| @@ -77,6 +77,10 @@ private[streaming] class ReceiverSupervisorImpl( | |||
| case CleanupOldBlocks(threshTime) => | |||
| logDebug("Received delete old batch signal") | |||
| cleanupOldBlocks(threshTime) | |||
| case RateLimitUpdate(eps) => { | |||
| blockGenerator.updateRate(eps.toInt) | |||
There was a problem hiding this comment.
Why the coercion from long to int?
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This supersedes #13, #11, #9 .