You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/07/25 23:33:12 UTC
[GitHub] [spark] JoshRosen opened a new pull request, #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
JoshRosen opened a new pull request, #37282:
URL: https://github.com/apache/spark/pull/37282
### What changes were proposed in this pull request?
This PR modifies `ExecutionListenerManager` so that its `ExecutionListenerBus` SparkListener is lazily registered during the first `.register(QueryExceutionListener)` (instead of eagerly registering it in the constructor).
### Why are the changes needed?
This addresses a ListenerBus performance problem in applications with large numbers of short-lived SparkSessions.
The `ExecutionListenerBus` SparkListener is unregistered by the ContextCleaner after its associated ExecutionListenerManager/SparkSession is garbage-collected (see #31839). If many sessions are rapidly created and destroyed but the driver GC doesn't run then this can result in large number of unused ExecutionListenerBus listeners being registered on the shared ListenerBus queue. This can cause performance problems in the ListenerBus because each listener invocation has some overhead.
In one real-world application with a very large driver heap and high rate of SparkSession creation (hundreds per minute), I saw 5000 idle ExecutionListenerBus listeners, resulting in ~50ms median event processing times on the shared listener queue.
This patch avoids this problem by making the listener registration lazy: if a short-lived SparkSession never uses QueryExecutionListeners then we won't register the ExecutionListenerBus and won't incur these overheads.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added a new unit test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon closed pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
HyukjinKwon closed pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
URL: https://github.com/apache/spark/pull/37282
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929497921
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +81,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
Review Comment:
ah, the problem is in `clone` ...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929505268
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +81,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
Review Comment:
I am not very familiar with the sql codebase - but I was assuming this was a common enough case ? If not, agree `lazy val` would be better here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929397395
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +83,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
+ // SPARK-39864: lazily create the listener bus on the first register() call in order to
+ // avoid listener overheads when QueryExecutionListeners aren't used:
+ @GuardedBy("this")
Review Comment:
I don't think there's a deadlock risk here (something that I always want to check whenever I'm adding synchronization to existing code).
The register(), unregister(), clear(), and listListeners() methods call `ListenerBus` methods and those methods are synchronized. However, I don't think they'll have circular waits because those methods are just updating ListenerBus data structures and aren't running code which could call back into ExecutionListenerManager.
I'm a bit less sure about clone(), though.
To play it safe, I'm going to see if I can remove the method-level `synchronized` and instead do something more surgical.
**Edit:** pushed a commit with a more conservative approach to synchronization.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #37282:
URL: https://github.com/apache/spark/pull/37282#issuecomment-1196279779
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929502561
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +81,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
Review Comment:
Updated in 030f6379b3c5c6419cf6a5cb165725e88774cc3f
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929397395
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +83,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
+ // SPARK-39864: lazily create the listener bus on the first register() call in order to
+ // avoid listener overheads when QueryExecutionListeners aren't used:
+ @GuardedBy("this")
Review Comment:
I don't think there's a deadlock risk here (something that I always want to check whenever I'm adding synchronization to existing code).
The register(), unregister(), clear(), and listListeners() methods call `ListenerBus` methods and those methods are synchronized. However, I don't think they'll have circular waits because those methods are just updating ListenerBus data structures and aren't running code which could call back into ExecutionListenerManager.
I'm a bit less sure about clone(), though.
To play it safe, I'm going to see if I can remove the method-level `synchronized` and instead do something more surgical.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929501950
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +81,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
Review Comment:
Yeah, I was worried about `clone`. Now that I think about it, though, I think a `lazy val` might be okay: most of the new `SparkSessions` will be child sessions of the root session, so the work I did in `clone` is only saving a registration of the root session and isn't saving any work in the child. Avoiding an unnecessary initialization in clone would only make a difference if you had a large number of child sessions which spawned large numbers of grandchildren sessions, but I'm not worried about that case. Let me update to make it a simple `lazy val`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] beliefer commented on pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #37282:
URL: https://github.com/apache/spark/pull/37282#issuecomment-1196161990
@JoshRosen Thank you for your overall thoughts on performance improvement.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929496679
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +81,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
Review Comment:
qq. can we just make this as a lazy val or a function?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] beliefer commented on pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
beliefer commented on PR #37282:
URL: https://github.com/apache/spark/pull/37282#issuecomment-1195097695
I have a question if one application need register many QueryExecutionListener, it seems cause performance fallback.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] mridulm commented on pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #37282:
URL: https://github.com/apache/spark/pull/37282#issuecomment-1194974570
> The fix look good @JoshRosen ! Could not find a clean way to remove `volatile` for `listenerBus` ... ah well.
This is obviously for the earlier change :-)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on PR #37282:
URL: https://github.com/apache/spark/pull/37282#issuecomment-1195799851
> I have a question if one application need register many QueryExecutionListener, it seems cause performance fallback.
Yes: if an application _doesn't_ use QueryExecutionListener then this PR will improve performance, but performance will be unchanged for applications where each SparkSession registers a QueryExecutionListener.
To improve performance in that case, we can consider restructuring the code so that only one listener is registered for each SparkContext. Internally, that singleton listener could maintain a hashmap from sessionUUID -> session-specific-listener-bus. This would mean that QueryExecutionListener event dispatch would be O(1) instead of O(numSessions), which is a huge improvement.
When implementing this idea, I'd like to avoid actual global singletons because they make testing harder. Instead, I'd probably have an `object ExecutionListenerManager { val registrationLock = new Object; def getOrRegisterListener(sc: SparkContext): ExecutionListenerManager }` where we have a global singleton lock which is used to register the per-SparkContext root listener manager.
I'd be interested in implementing that larger change, but I'd like to do it in a separate PR because it's a lot larger in scope. This current PR's fix is surgical and improves performance in a real-world use-case, so I'd like to start by merging this small fix and tackle the larger redesign in a separate followup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] JoshRosen commented on a diff in pull request #37282: [SPARK-39864][SQL] Lazily register ExecutionListenerBus
Posted by GitBox <gi...@apache.org>.
JoshRosen commented on code in PR #37282:
URL: https://github.com/apache/spark/pull/37282#discussion_r929510207
##########
sql/core/src/main/scala/org/apache/spark/sql/util/QueryExecutionListener.scala:
##########
@@ -81,7 +81,10 @@ class ExecutionListenerManager private[sql](
loadExtensions: Boolean)
extends Logging {
- private val listenerBus = new ExecutionListenerBus(this, session)
Review Comment:
Thinking about this some more, I don't think the `root -> child -> grandchild` session case will be common in Spark itself (I think the ThriftServer would only create one layer of child sessions), but that scenario might crop up in applications built on Spark. Given this, I'm going to roll back to what I had earlier with the `@volatile` Option and the initialization lock. It's a _bit_ ugly but not _that_ ugly and provides some additional benefit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org