Propagate thread locals to Delta thread pools #2154
Closed
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
larsk-db
reviewed
Oct 9, 2023
| // Capture an immutable threadsafe snapshot of the current local properties | ||
| val capturedProperties = sparkContext | ||
| .map(sc => CapturedSparkThreadLocals.toValuesArray( | ||
| org.apache.spark.util.Utils.cloneProperties(sc.getLocalProperties))) |
There was a problem hiding this comment.
import Utils instead of using qualified name here? You can rename it (e.g. to SparkUtils) if it clashes.
Comment on lines
89
to
91
| capturedProperties.foreach { p => | ||
| sparkContext.foreach(_.setLocalProperties(CapturedSparkThreadLocals.toProperties(p))) | ||
| } |
There was a problem hiding this comment.
Suggested change
| capturedProperties.foreach { p => | |
| sparkContext.foreach(_.setLocalProperties(CapturedSparkThreadLocals.toProperties(p))) | |
| } | |
| for { | |
| p <- capturedProperties | |
| sc <- sparkContext | |
| } sc.setLocalProperties(CapturedSparkThreadLocals.toProperties(p)) |
Isn't this much more readable? ;)
| throw t | ||
| } finally { | ||
| TaskContext.setTaskContext(previousTaskContext) | ||
| previousProperties.foreach(p => sparkContext.foreach(_.setLocalProperties(p))) |
Comment on lines
115
to
117
| props.foreach { kvp => | ||
| resultProps.put(kvp._1, kvp._2) | ||
| } |
There was a problem hiding this comment.
Or, hear me out:
Suggested change
| props.foreach { kvp => | |
| resultProps.put(kvp._1, kvp._2) | |
| } | |
| for ((key, value) <- props) { | |
| resultProps.put(key, value) | |
| } |
| } | ||
|
|
||
| test("That CapturedSparkThreadLocals properly restores the existing spark properties." + | ||
| " Changes to local properties inside a task do not affect the original properties") { |
fred-db
force-pushed
the
propagate-threadlocals
branch
from
60faa14
to
da5724d
Compare
|
Hi @larsk-db , thank you so much for reviewing! :) I addressed your comments, hope it looks good now. I also moved the |
fred-db
force-pushed
the
propagate-threadlocals
branch
from
da5724d
to
0d69c60
Compare
fred-db
changed the title
5 tasks
5 tasks
xupefei
pushed a commit
to xupefei/delta
that referenced
this pull request
Oct 31, 2023
* The default thread pool executor in Apache Spark does not forward thread locals to threads spawned in a thread pool. * This can cause issues if the threads depend on the thread locals. * To fix this, we introduce a wrapper class around the thread pool executor that forwards thread locals. Closes delta-io#2154 GitOrigin-RevId: 9e9423e4b041232457ffaab18f5f96490bb45b88
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.
Which Delta project/connector is this regarding?
Description
How was this patch tested?
SparkThreadLocalForwardingExecutorto ensure thread locals are forwarded and reset after future finished.Does this PR introduce any user-facing changes?
No