venkata91 commented on code in PR #58437:
URL: https://github.com/apache/spark/pull/58437#discussion_r3997003774
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4199,7 +4200,10 @@ private[spark] class DAGScheduler(
execId = execId,
fileLost = fileLost,
hostToUnregisterOutputs = workerHost,
- maybeEpoch = None)
+ maybeEpoch = None,
+ // Executor loss (not a fetch failure): preserve shuffles whose output
is reliably stored
+ // off-executor. Their data survives the executor, so recomputing them
would be wasteful.
+ skipReliablyStored = true)
Review Comment:
Good repro, thanks. Fixed: `shuffleFileLostEpoch(execId)` is now stamped
only when the cleanup was complete (`!preservedReliable`), so a selective
executor-loss cleanup that preserved reliable output no longer suppresses a
later same-epoch FetchFailed. The FetchFailed bulk cleanup also passes
`respectReliablyStored = true` now, so it removes non-reliable output only.
Covered by `SPARK-59138: same-epoch FetchFailed after selective executor-loss
cleanup is not skipped` and a new mixed reliable/local test.
##########
core/src/main/scala/org/apache/spark/shuffle/ShuffleHandle.scala:
##########
@@ -25,4 +25,12 @@ import org.apache.spark.annotation.DeveloperApi
* @param shuffleId ID of the shuffle
*/
@DeveloperApi
-abstract class ShuffleHandle(val shuffleId: Int) extends Serializable {}
+abstract class ShuffleHandle(val shuffleId: Int) extends Serializable {
+ /**
+ * Whether this shuffle's output is stored reliably outside the executors
that produced it (e.g.
+ * a remote shuffle service). When true, losing an executor does not lose
this shuffle's output,
+ * so its map outputs are not unregistered on executor loss. Defaults to
false; a ShuffleManager
Review Comment:
Updated the contract to require host-independence: "stored reliably, off the
executor that produced it ... losing the executor or its host does not lose the
output". That matches `ShuffleDriverComponents.supportsReliableStorage()` and
rules out same-host another-process storage.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1228,8 +1228,20 @@ private[spark] class TaskSetManager(
// pipelined set and aborts the whole group. Note isZombie already skips a
fully-complete
// producer's set; this guard also covers a PARTIALLY-complete producer
losing an executor on
// decommission.
+
+ // OR, not AND: a shuffle reliably stored off-executor (globally, or just
this one via a remote
+ // shuffle service) keeps its map output when the executor dies. The
per-shuffle bit only ever
+ // adds reliability (defaults to false, set true solely by an opting-in
manager), so a false
+ // there means "no info", not "unreliable".
+ val reliablyStored =
sched.sc.shuffleDriverComponents.supportsReliableStorage() ||
Review Comment:
Made it authoritative. `TaskSetManager` now reads only
`sched.mapOutputTracker.isReliablyStored(shuffleId)`, which is the per-shuffle
value resolved once at `registerShuffle`
(`handle.reliablyStored.getOrElse(global)`). The old global-flag branch is
gone, so a `Some(false)` fallback shuffle under a `true` global correctly
reruns and cleans up.
##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -1112,11 +1123,38 @@ class DAGSchedulerSuite extends SparkFunSuite with
TempLocalSparkContext with Ti
completeShuffleMapStageSuccessfully(0, 0, 1)
runEvent(ExecutorLost("hostA-exec", event))
verify(blockManagerMaster, times(1)).removeExecutorAsync("hostA-exec")
- verify(mapOutputTracker, times(0)).removeOutputsOnExecutor("hostA-exec")
+ verify(mapOutputTracker, times(0)).removeOutputsOnExecutor("hostA-exec",
true)
assert(mapOutputTracker.getMapSizesByExecutorId(shuffleId,
0).map(_._1).toSet ===
HashSet(makeBlockManagerId("hostA"), makeBlockManagerId("hostB")))
}
+ test("SPARK-59138: executor loss keeps a reliably-stored shuffle but drops a
local-disk one") {
+ // No external shuffle service, so a plain (local-disk) shuffle's outputs
are lost on executor
+ // loss, but a per-shuffle reliably-stored one survives.
+ conf.set(config.SHUFFLE_SERVICE_ENABLED.key, "false")
Review Comment:
Added: `SPARK-59138: per-shuffle reliablyStored=false overrides a global
supportsReliableStorage` sets `supportsReliableStorage() = true` with one
handle reporting `Some(false)`, and asserts that shuffle's output is dropped on
executor loss while the reliable one is preserved.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1228,8 +1228,13 @@ private[spark] class TaskSetManager(
// pipelined set and aborts the whole group. Note isZombie already skips a
fully-complete
// producer's set; this guard also covers a PARTIALLY-complete producer
losing an executor on
// decommission.
+
+ // The tracker's stored value already folds the app-global flag and the
per-shuffle handle
+ // together at registration (see DAGScheduler.createShuffleMapStage), and
a shuffle map task set
+ // always has its shuffle registered before it is submitted. Read it as
the single source.
+ val reliablyStored =
taskSet.shuffleId.exists(sched.mapOutputTracker.isReliablyStored)
val maybeShuffleMapOutputLoss = isShuffleMapTasks && !taskSet.isPipelined
&&
- !sched.sc.shuffleDriverComponents.supportsReliableStorage() &&
+ !reliablyStored &&
Review Comment:
Added: `SPARK-59138: task reruns on executor lost for an explicitly
unreliable shuffle` in TaskSetManagerSuite registers the shuffle with
`isReliablyStored = false`, completes a map task on the executor, loses it, and
asserts the task is invalidated (`!successful`).
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1044,20 +1059,48 @@ private[spark] class MapOutputTrackerMaster(
/**
* Removes all shuffle outputs associated with this host. Note that this
will also remove
* outputs which are served by an external shuffle server (if one exists).
+ *
+ * When `respectReliablyStored` is true (executor/worker loss rather than a
fetch failure),
Review Comment:
Done. Moved the selective contract onto the Boolean overloads and marked the
one-arg wrappers as unconditional cleanup.
##########
core/src/main/scala/org/apache/spark/StreamingShuffleOutputTracker.scala:
##########
@@ -38,7 +38,12 @@ import org.apache.spark.util.ThreadUtils
*/
private[spark] trait ShuffleOutputTrackerMaster {
/** Register a shuffle so its outputs can be tracked. `jobId` is used by the
streaming tracker. */
- def registerShuffle(shuffleId: Int, numMaps: Int, numReduces: Int, jobId:
Int): Unit
+ def registerShuffle(
+ shuffleId: Int,
+ numMaps: Int,
+ numReduces: Int,
+ jobId: Int,
+ isReliablyStored: Boolean = false): Unit
Review Comment:
Done. Documented the split on the common `registerShuffle`:
`isReliablyStored` is honored only by `MapOutputTrackerMaster`; the streaming
tracker ignores it because pipelined output is located/invalidated via the
streaming task-location registry, not map-status cleanup.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4190,16 +4193,18 @@ private[spark] class DAGScheduler(
private[scheduler] def handleExecutorLost(
execId: String,
workerHost: Option[String]): Unit = {
- // if the cluster manager explicitly tells us that the entire worker was
lost, then
- // we know to unregister shuffle output. (Note that "worker" specifically
refers to the process
- // from a Standalone cluster, where the shuffle service lives in the
Worker.)
- val fileLost = !sc.shuffleDriverComponents.supportsReliableStorage() &&
- (workerHost.isDefined || !env.blockManager.externalShuffleServiceEnabled)
+ // Whether these outputs are candidates for removal at all; reliability is
then honored per
+ // shuffle via respectReliablyStored below. workerHost.isDefined means the
whole Standalone
+ // worker (which hosts the shuffle service) is gone.
+ val fileLost = workerHost.isDefined ||
!env.blockManager.externalShuffleServiceEnabled
removeExecutorAndUnregisterOutputs(
execId = execId,
fileLost = fileLost,
hostToUnregisterOutputs = workerHost,
- maybeEpoch = None)
+ maybeEpoch = None,
+ // Executor loss (not a fetch failure): preserve shuffles whose output
is reliably stored
Review Comment:
Done. Qualified the Scaladoc to say only non-reliably-stored output is
treated as lost on executor/worker loss, and added `@param` entries for
`ignoreShuffleFileLostEpoch` and `respectReliablyStored`.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -4267,7 +4272,8 @@ private[spark] class DAGScheduler(
fileLost: Boolean,
hostToUnregisterOutputs: Option[String],
maybeEpoch: Option[Long] = None,
- ignoreShuffleFileLostEpoch: Boolean = false): Unit = {
+ ignoreShuffleFileLostEpoch: Boolean = false,
+ respectReliablyStored: Boolean = false): Unit = {
Review Comment:
Done. Qualified the Scaladoc to say only non-reliably-stored output is
treated as lost on executor/worker loss, and added `@param` entries for
`ignoreShuffleFileLostEpoch` and `respectReliablyStored`.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
