[SPARK-28625][Core] Indeterminate shuffle support in Shuffle Writer API #25361
Conversation
Add indeterminate stage rerun support in shuffle writer api (cherry picked from commit 99c2b4a) Signed-off-by: Yuanjian Li <[email protected]>
xuanyuanking
changed the title
|
Test build #108662 has finished for PR 25361 at commit
|
|
Test build #108663 has finished for PR 25361 at commit
|
| @@ -39,16 +39,20 @@ | |||
| /** | |||
| * Called once per map task to create a writer that will be responsible for persisting all the | |||
| * partitioned bytes written by that map task. | |||
| * @param shuffleId Unique identifier for the shuffle the map task is a part of | |||
| * @param shuffleId Unique identifier for the shuffle the map task is a part of | |||
| * @param shuffleGenerationId The shuffle generation ID of the stage that this task belongs to, | |||
There was a problem hiding this comment.
Why can't we just use the mapTaskAttemptId? (In fact I wonder if we can just remove shuffleId and mapId and just use mapTaskAttemptId as a global identifier, but that might be a bit ambitious.)
There was a problem hiding this comment.
Specifically taking a look at the linked PR for indeterminate retries - I'd expect that on a rolled back map stage, the implementation of this plugin will be given a different mapTaskAttemptId anyways since that's going to be updated on the resubmit. So, we'll end up opening a new writer regardless, but, we could have gotten the same behavior just by using the mapTaskAttemptId.
There was a problem hiding this comment.
Actually let's move discussion over to #24892
There was a problem hiding this comment.
Yep, thanks for your faster review, I describe the requirement in #24892 (comment)
| ShuffleMapOutputWriter mapOutputWriter = shuffleExecutorComponents | ||
| .createMapOutputWriter(shuffleId, mapId, mapTaskAttemptId, numPartitions); | ||
| ShuffleMapOutputWriter mapOutputWriter = shuffleExecutorComponents.createMapOutputWriter( | ||
| shuffleId, -1, mapId, mapTaskAttemptId, numPartitions); |
There was a problem hiding this comment.
Why are we always passing in -1 here?
There was a problem hiding this comment.
This PR created for quick reviewing of API changes, you can see the real scenario of shuffleGeneraionId here :)
|
Close this preview PR, the API changed in #25620, which uses map task attempt Id as the map id during writer creation. |
What changes were proposed in this pull request?
Add indeterminate stage rerun support in shuffle writer API, the usage of newly added
shuffleGenerationIdparam is in #24892.How was this patch tested?
Existing UT.