[SPARK-19182][DStream] Optimize the lock in StreamingJobProgressListener to not block UI when generating Streaming jobs #16601
Conversation
This file contains 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
…n generating Streaming jobs
|
Test build #71441 has finished for PR 16601 at commit
|
zsxwing
requested changes
Jan 17, 2017
| jobOption.foreach(_.setCallSite(outputStream.creationSite)) | ||
| jobOption | ||
| } | ||
| val jobs = getOutputStreams().flatMap { outputStream => |
There was a problem hiding this comment.
synchronized is to make sure writeObject never write some intermediate states of DStreamGraph.
There was a problem hiding this comment.
Yes, I have put the question to be too simple
|
Test build #71492 has finished for PR 16601 at commit
|
|
@zsxwing Take a review please! |
zsxwing
approved these changes
Jan 18, 2017
| var rememberDuration: Duration = null | ||
| var checkpointInProgress = false | ||
|
|
||
| var zeroTime: Time = null | ||
| var startTime: Time = null | ||
| var batchDuration: Duration = null | ||
| var numReceivers: Int = 0 |
| @@ -31,12 +31,15 @@ final private[streaming] class DStreamGraph extends Serializable with Logging { | |||
| private val inputStreams = new ArrayBuffer[InputDStream[_]]() | |||
| private val outputStreams = new ArrayBuffer[DStream[_]]() | |||
|
|
|||
| val inputStreamNameAndID = new ArrayBuffer[(String, Int)]() | |||
There was a problem hiding this comment.
nit: change it to @volatile private var inputStreamNameAndID: Seq[(String, Int)] = Nil and just set it in start. Don't expose a mutable ArrayBuffer to the caller.
| def getInputStreamName(streamId: Int): Option[String] = synchronized { | ||
| inputStreams.find(_.id == streamId).map(_.name) | ||
| } | ||
| def getReceiverNumber: Int = numReceivers |
There was a problem hiding this comment.
nit getNumReceivers for consistence.
| @@ -197,17 +197,17 @@ private[spark] class StreamingJobProgressListener(ssc: StreamingContext) | |||
| } | |||
|
|
|||
| def retainedCompletedBatches: Seq[BatchUIData] = synchronized { | |||
| completedBatchUIData.toSeq | |||
| completedBatchUIData | |||
There was a problem hiding this comment.
nit: could you change toSeq to toIndexedSeq to avoid exposing a mutable collection.
|
Test build #71553 has finished for PR 16601 at commit
|
|
LGTM. Merging to master. Thanks! |
zzcclp
added a commit
to zzcclp/spark
that referenced
this pull request
Jan 20, 2017
…Listener to not block UI when generating Streaming jobs apache#16601
uzadude
pushed a commit
to uzadude/spark
that referenced
this pull request
Jan 27, 2017
…ner to not block UI when generating Streaming jobs ## What changes were proposed in this pull request? When DStreamGraph is generating a job, it will hold a lock and block other APIs. Because StreamingJobProgressListener (numInactiveReceivers, streamName(streamId: Int), streamIds) needs to call DStreamGraph's methods to access some information, the UI may hang if generating a job is very slow (e.g., talking to the slow Kafka cluster to fetch metadata). It's better to optimize the locks in DStreamGraph and StreamingJobProgressListener to make the UI not block by job generation. ## How was this patch tested? existing ut cc zsxwing Author: uncleGen <[email protected]> Closes apache#16601 from uncleGen/SPARK-19182.
cmonkey
pushed a commit
to cmonkey/spark
that referenced
this pull request
Feb 15, 2017
…ner to not block UI when generating Streaming jobs ## What changes were proposed in this pull request? When DStreamGraph is generating a job, it will hold a lock and block other APIs. Because StreamingJobProgressListener (numInactiveReceivers, streamName(streamId: Int), streamIds) needs to call DStreamGraph's methods to access some information, the UI may hang if generating a job is very slow (e.g., talking to the slow Kafka cluster to fetch metadata). It's better to optimize the locks in DStreamGraph and StreamingJobProgressListener to make the UI not block by job generation. ## How was this patch tested? existing ut cc zsxwing Author: uncleGen <[email protected]> Closes apache#16601 from uncleGen/SPARK-19182.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
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.
What changes were proposed in this pull request?
When DStreamGraph is generating a job, it will hold a lock and block other APIs. Because StreamingJobProgressListener (numInactiveReceivers, streamName(streamId: Int), streamIds) needs to call DStreamGraph's methods to access some information, the UI may hang if generating a job is very slow (e.g., talking to the slow Kafka cluster to fetch metadata).
It's better to optimize the locks in DStreamGraph and StreamingJobProgressListener to make the UI not block by job generation.
How was this patch tested?
existing ut
cc @zsxwing