[SPARK-35654][CORE] Allow ShuffleDataIO control DiskBlockManager.deleteFilesOnStop #32784
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
Member
Author
|
Kubernetes integration test unable to build dist. exiting with code: 1 |
|
Test build #139339 has finished for PR 32784 at commit
|
Member
|
cc @Ngone51 too FYI |
mridulm
approved these changes
Jun 6, 2021
viirya
approved these changes
Jun 6, 2021
Member
Author
|
Thank you so much, @mridulm , @viirya , @HyukjinKwon . |
flyrain
pushed a commit
to flyrain/spark
that referenced
this pull request
Sep 21, 2021
…teFilesOnStop
### What changes were proposed in this pull request?
This PR aims to change `DiskBlockManager` like the following to allow `ShuffleDataIO` to decide the behavior of shuffle file deletion.
```scala
- private[spark] class DiskBlockManager(conf: SparkConf, deleteFilesOnStop: Boolean)
+ private[spark] class DiskBlockManager(conf: SparkConf, var deleteFilesOnStop: Boolean)
```
### Why are the changes needed?
`SparkContext` creates
1. `SparkEnv` (with `BlockManager` and its `DiskBlockManager`)
2. loads `ShuffleDataIO`
3. initialize block manager.
```scala
_env = createSparkEnv(_conf, isLocal, listenerBus)
...
_shuffleDriverComponents = ShuffleDataIOUtils.loadShuffleDataIO(config).driver()
_shuffleDriverComponents.initializeApplication().asScala.foreach { case (k, v) =>
_conf.set(ShuffleDataIOUtils.SHUFFLE_SPARK_CONF_PREFIX + k, v)
}
...
_env.blockManager.initialize(_applicationId)
...
```
`DiskBlockManager` is created first at `BlockManager` constructor and we cannot change `deleteFilesOnStop` later at `ShuffleDataIO`. By switching to `var`, we can implement enhanced shuffle data management feature via `ShuffleDataIO` like apache#32730 .
```
val diskBlockManager = {
// Only perform cleanup if an external service is not serving our shuffle files.
val deleteFilesOnStop =
!externalShuffleServiceEnabled || executorId == SparkContext.DRIVER_IDENTIFIER
new DiskBlockManager(conf, deleteFilesOnStop)
}
```
### Does this PR introduce _any_ user-facing change?
No. This is a private class.
### How was this patch tested?
N/A
Closes apache#32784 from dongjoon-hyun/SPARK-35654.
Authored-by: Dongjoon Hyun <dhyun@apple.com>
Signed-off-by: Dongjoon Hyun <dhyun@apple.com>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
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?
This PR aims to change
DiskBlockManagerlike the following to allowShuffleDataIOto decide the behavior of shuffle file deletion.Why are the changes needed?
SparkContextcreatesSparkEnv(withBlockManagerand itsDiskBlockManager)ShuffleDataIODiskBlockManageris created first atBlockManagerconstructor and we cannot changedeleteFilesOnStoplater atShuffleDataIO. By switching tovar, we can implement enhanced shuffle data management feature viaShuffleDataIOlike #32730 .Does this PR introduce any user-facing change?
No. This is a private class.
How was this patch tested?
N/A