[SPARK-59138][CORE] Preserve reliably-stored shuffle map outputs on executor loss - #58437
[SPARK-59138][CORE] Preserve reliably-stored shuffle map outputs on executor loss#58437venkata91 wants to merge 1 commit into
Conversation
16721fb to
cbd2ad3
Compare
venkata91
changed the title
cbd2ad3 to
7c2854b
Compare
7c2854b to
4350c69
Compare
uros-b
left a comment
uros-b
left a comment
There was a problem hiding this comment.
Left a comment here, but please also ping committers with deeper shuffle and executor-loss expertise, for further PR review
| status.removeOutputsOnHost(host) | ||
| } | ||
| } | ||
| incrementEpoch() |
There was a problem hiding this comment.
removeOutputsOnHost/removeOutputsOnExecutor unconditionally call incrementEpoch() even when skipReliablyStored = true and every shuffle is reliably stored (nothing was actually removed from the tracker). The epoch bump causes every executor worker to invalidate its cached map-output info and issue a round-trip to the driver on next use, even though the tracker state is unchanged. In the mixed-configuration scenario this optimization targets (remote shuffle service with frequent spot-instance turnover) this produces spurious cache churn on every executor loss. Fix: track whether any status's removal call changed state and guard incrementEpoch() accordingly.
There was a problem hiding this comment.
Addressed it. Yes, we can avoid the incrementEpoch if all the shuffles are reliably stored during an executor loss.
Thanks for your review. Long time back, I have worked on this layer for push-based-shuffle. It is been a while now, looks like lot has changed. :)
…xecutor loss On executor loss the DAGScheduler unregisters all map outputs on that executor, forcing a map-stage recompute. For shuffles whose output is reliably stored off-executor (e.g. a remote shuffle service like Celeborn), the data survives the executor, so the recompute is wasteful. Reliability is per-shuffle, not app-global: under a mixed/fallback setup one shuffle can live on the remote service while another falls back to local disk on the same executor. Carry the bit on ShuffleHandle.isReliablyStored, store it per shuffle in MapOutputTracker, and skip only reliably-stored shuffles on executor/worker loss. Genuine fetch failures still unregister everything.
4350c69 to
c499375
Compare
2 participants
What changes were proposed in this pull request?
On executor loss,
DAGSchedulerunregisters all map outputs on that executor and forces a map-stage recompute. For shuffles whose output is reliably stored off-executor (e.g. a remote shuffle service such as Celeborn), the data survives the executor, so the recompute is wasteful.Reliability is per-shuffle, not app-global.
ShuffleDriverComponents.supportsReliableStorage()is a single application-wide flag, but under a mixed/fallback setup one shuffle can live on the remote service while another falls back to local disk on the same executor.This PR makes reliability per-shuffle:
ShuffleHandle.isReliablyStored(defaultfalse), overridable by aShuffleManagerthat routes a shuffle to reliable storage.MapOutputTrackerstores the bit per shuffle (ShuffleStatus), exposesisReliablyStored(shuffleId), andremoveOutputsOnExecutor/removeOutputsOnHostgain askipReliablyStoredoverload that leaves reliably-stored shuffles registered.DAGSchedulerpassesskipReliablyStored = trueon executor loss (handleExecutorLost) and worker loss (handleWorkerRemoved); the FetchFailed path keepsfalseso a genuine fetch failure still unregisters everything.TaskSetManager.executorLosthonors per-shuffle reliability in its re-run gate.Why are the changes needed?
With a remote shuffle service in a mixed configuration, losing an executor needlessly recomputes map stages whose output is safely stored on the service. The existing app-global flag cannot express "this shuffle is reliable but that one is not," so it either recomputes reliable shuffles or skips recompute for local-disk fallback shuffles that were genuinely lost.
Does this PR introduce any user-facing change?
No.
ShuffleHandle.isReliablyStoreddefaults tofalse, so behavior is unchanged unless aShuffleManageropts in.How was this patch tested?
MapOutputTrackerSuitetest covering a mixed reliable / local-disk scenario: executor loss preserves the reliably-stored shuffle and drops the local-disk one, while a fetch failure removes both.DAGSchedulerSuiteverifications for the executor-loss vs fetch-failureskipReliablyStoredwiring.DAGSchedulerSuite,MapOutputTrackerSuite, andTaskSetManagerSuitepass; scalastyle clean.Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)