You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "ivoson (via GitHub)" <gi...@apache.org> on 2025/12/02 01:31:32 UTC
[PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
ivoson opened a new pull request, #53274:
URL: https://github.com/apache/spark/pull/53274
### What changes were proposed in this pull request?
Rollback shuffle map stages when shuffle checksum mismatch detected:
- cancel and resubmit the stage if it's running;
- clean up the shuffle status to ensure it'll be resubmitted;
- mark rollback attemptId and ignore the results from these elder attempts which may consume inconsistent data;
### Why are the changes needed?
To ensure all the succeeding stages will be re-submitted and fully-retry when there is shuffle checksum mismatch detected.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
UT added.
### Was this patch authored or co-authored using generative AI tooling?
No
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #53274: [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected
URL: https://github.com/apache/spark/pull/53274
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2610302440
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2396,7 +2507,7 @@ private[spark] class DAGScheduler(
* @param stagesToRollback stages to roll back
* @return Shuffle map stages which need and can be rolled back
*/
- private def abortStageWithInvalidRollBack(stagesToRollback: HashSet[Stage]): HashSet[Stage] = {
+ private def abortStagesUnableToRollback(stagesToRollback: HashSet[Stage]): HashSet[Stage] = {
Review Comment:
how about `filterAndAbortUnrollbackableStages`? It returns rollbackable stages which is like `.filter`.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2610307715
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1560,42 +1560,27 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if !sms.isAvailable =>
- val needFullStageRetry = if (sms.shuffleDep.checksumMismatchFullRetryEnabled) {
- // When the parents of this stage are indeterminate (e.g., some parents are not
- // checkpointed and checksum mismatches are detected), the output data of the parents
- // may have changed due to task retries. For correctness reason, we need to
- // retry all tasks of the current stage. The legacy way of using current stage's
- // deterministic level to trigger full stage retry is not accurate.
- stage.isParentIndeterminate
- } else {
- if (stage.isIndeterminate) {
- // already executed at least once
- if (sms.getNextAttemptId > 0) {
- // While we previously validated possible rollbacks during the handling of a FetchFailure,
- // where we were fetching from an indeterminate source map stages, this later check
- // covers additional cases like recalculating an indeterminate stage after an executor
- // loss. Moreover, because this check occurs later in the process, if a result stage task
- // has successfully completed, we can detect this and abort the job, as rolling back a
- // result stage is not possible.
- val stagesToRollback = collectSucceedingStages(sms)
- abortStageWithInvalidRollBack(stagesToRollback)
- // stages which cannot be rolled back were aborted which leads to removing the
- // the dependant job(s) from the active jobs set
- val numActiveJobsWithStageAfterRollback =
- activeJobs.count(job => stagesToRollback.contains(job.finalStage))
- if (numActiveJobsWithStageAfterRollback == 0) {
- logInfo(log"All jobs depending on the indeterminate stage " +
- log"(${MDC(STAGE_ID, stage.id)}) were aborted so this stage is not needed anymore.")
- return
- }
+ if (!sms.shuffleDep.checksumMismatchFullRetryEnabled && stage.isIndeterminate) {
Review Comment:
is it possible to merge the old indeterminate stage code path into the new framework? I think this is just a special case where we know checksum always mismatch.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #53274:
URL: https://github.com/apache/spark/pull/53274#issuecomment-3654646736
gentle ping @mridulm in case you missed this
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2610064628
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1913,16 +1898,127 @@ private[spark] class DAGScheduler(
/**
* If a map stage is non-deterministic, the map tasks of the stage may return different result
- * when re-try. To make sure data correctness, we need to re-try all the tasks of its succeeding
- * stages, as the input data may be changed after the map tasks are re-tried. For stages where
- * rollback and retry all tasks are not possible, we will need to abort the stages.
+ * when re-try. To make sure data correctness, we need to clean up shuffles to make sure succeeding
+ * stages will be resubmitted and re-try all the tasks, as the input data may be changed after
+ * the map tasks are re-tried. For stages where rollback and retry all tasks are not possible,
+ * we will need to abort the stages.
+ */
+ private[scheduler] def rollbackSucceedingStages(mapStage: ShuffleMapStage): Unit = {
+ val stagesToRollback = collectSucceedingStages(mapStage).filterNot(_ == mapStage)
+ val stagesCanRollback = abortStagesUnableToRollback(stagesToRollback)
Review Comment:
```suggestion
val rollingBackStages = abortStagesUnableToRollback(stagesToRollback)
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2610066723
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1913,16 +1898,127 @@ private[spark] class DAGScheduler(
/**
* If a map stage is non-deterministic, the map tasks of the stage may return different result
- * when re-try. To make sure data correctness, we need to re-try all the tasks of its succeeding
- * stages, as the input data may be changed after the map tasks are re-tried. For stages where
- * rollback and retry all tasks are not possible, we will need to abort the stages.
+ * when re-try. To make sure data correctness, we need to clean up shuffles to make sure succeeding
+ * stages will be resubmitted and re-try all the tasks, as the input data may be changed after
+ * the map tasks are re-tried. For stages where rollback and retry all tasks are not possible,
+ * we will need to abort the stages.
+ */
+ private[scheduler] def rollbackSucceedingStages(mapStage: ShuffleMapStage): Unit = {
+ val stagesToRollback = collectSucceedingStages(mapStage).filterNot(_ == mapStage)
+ val stagesCanRollback = abortStagesUnableToRollback(stagesToRollback)
Review Comment:
to match https://github.com/apache/spark/pull/53274/files?diff=unified&w=1#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6R2317
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2610075582
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2206,7 +2313,11 @@ private[spark] class DAGScheduler(
// guaranteed to be determinate, so the input data of the reducers will not change
// even if the map tasks are re-tried.
if (mapStage.isIndeterminate && !mapStage.shuffleDep.checksumMismatchFullRetryEnabled) {
- abortUnrollbackableStages(mapStage)
+ val stagesToRollback = collectSucceedingStages(mapStage)
+ val rollingBackStages = abortStagesUnableToRollback(stagesToRollback)
Review Comment:
```suggestion
val stagesCanRollback = abortStagesUnableToRollback(stagesToRollback)
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #53274:
URL: https://github.com/apache/spark/pull/53274#issuecomment-3675600323
thanks, merging to master/4.1!
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2610064628
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1913,16 +1898,127 @@ private[spark] class DAGScheduler(
/**
* If a map stage is non-deterministic, the map tasks of the stage may return different result
- * when re-try. To make sure data correctness, we need to re-try all the tasks of its succeeding
- * stages, as the input data may be changed after the map tasks are re-tried. For stages where
- * rollback and retry all tasks are not possible, we will need to abort the stages.
+ * when re-try. To make sure data correctness, we need to clean up shuffles to make sure succeeding
+ * stages will be resubmitted and re-try all the tasks, as the input data may be changed after
+ * the map tasks are re-tried. For stages where rollback and retry all tasks are not possible,
+ * we will need to abort the stages.
+ */
+ private[scheduler] def rollbackSucceedingStages(mapStage: ShuffleMapStage): Unit = {
+ val stagesToRollback = collectSucceedingStages(mapStage).filterNot(_ == mapStage)
+ val stagesCanRollback = abortStagesUnableToRollback(stagesToRollback)
Review Comment:
```suggestion
val rollingBackStages = abortStagesUnableToRollback(stagesToRollback)
```
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1913,16 +1898,127 @@ private[spark] class DAGScheduler(
/**
* If a map stage is non-deterministic, the map tasks of the stage may return different result
- * when re-try. To make sure data correctness, we need to re-try all the tasks of its succeeding
- * stages, as the input data may be changed after the map tasks are re-tried. For stages where
- * rollback and retry all tasks are not possible, we will need to abort the stages.
+ * when re-try. To make sure data correctness, we need to clean up shuffles to make sure succeeding
+ * stages will be resubmitted and re-try all the tasks, as the input data may be changed after
+ * the map tasks are re-tried. For stages where rollback and retry all tasks are not possible,
+ * we will need to abort the stages.
+ */
+ private[scheduler] def rollbackSucceedingStages(mapStage: ShuffleMapStage): Unit = {
+ val stagesToRollback = collectSucceedingStages(mapStage).filterNot(_ == mapStage)
+ val stagesCanRollback = abortStagesUnableToRollback(stagesToRollback)
Review Comment:
to match https://github.com/apache/spark/pull/53274/files?diff=unified&w=1#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6R2317
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on PR #53274:
URL: https://github.com/apache/spark/pull/53274#issuecomment-3631036543
cc @cloud-fan @mridulm
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2610064628
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1913,16 +1898,127 @@ private[spark] class DAGScheduler(
/**
* If a map stage is non-deterministic, the map tasks of the stage may return different result
- * when re-try. To make sure data correctness, we need to re-try all the tasks of its succeeding
- * stages, as the input data may be changed after the map tasks are re-tried. For stages where
- * rollback and retry all tasks are not possible, we will need to abort the stages.
+ * when re-try. To make sure data correctness, we need to clean up shuffles to make sure succeeding
+ * stages will be resubmitted and re-try all the tasks, as the input data may be changed after
+ * the map tasks are re-tried. For stages where rollback and retry all tasks are not possible,
+ * we will need to abort the stages.
+ */
+ private[scheduler] def rollbackSucceedingStages(mapStage: ShuffleMapStage): Unit = {
+ val stagesToRollback = collectSucceedingStages(mapStage).filterNot(_ == mapStage)
+ val stagesCanRollback = abortStagesUnableToRollback(stagesToRollback)
Review Comment:
```suggestion
val stagesRollingBack = abortStagesUnableToRollback(stagesToRollback)
```
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-54556][CORE] Rollback succeeding shuffle map stages when shuffle checksum mismatch detected [spark]
Posted by "ivoson (via GitHub)" <gi...@apache.org>.
ivoson commented on code in PR #53274:
URL: https://github.com/apache/spark/pull/53274#discussion_r2618564078
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2396,7 +2507,7 @@ private[spark] class DAGScheduler(
* @param stagesToRollback stages to roll back
* @return Shuffle map stages which need and can be rolled back
*/
- private def abortStageWithInvalidRollBack(stagesToRollback: HashSet[Stage]): HashSet[Stage] = {
+ private def abortStagesUnableToRollback(stagesToRollback: HashSet[Stage]): HashSet[Stage] = {
Review Comment:
Sounds good. Updated. Thanks.
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1560,42 +1560,27 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if !sms.isAvailable =>
- val needFullStageRetry = if (sms.shuffleDep.checksumMismatchFullRetryEnabled) {
- // When the parents of this stage are indeterminate (e.g., some parents are not
- // checkpointed and checksum mismatches are detected), the output data of the parents
- // may have changed due to task retries. For correctness reason, we need to
- // retry all tasks of the current stage. The legacy way of using current stage's
- // deterministic level to trigger full stage retry is not accurate.
- stage.isParentIndeterminate
- } else {
- if (stage.isIndeterminate) {
- // already executed at least once
- if (sms.getNextAttemptId > 0) {
- // While we previously validated possible rollbacks during the handling of a FetchFailure,
- // where we were fetching from an indeterminate source map stages, this later check
- // covers additional cases like recalculating an indeterminate stage after an executor
- // loss. Moreover, because this check occurs later in the process, if a result stage task
- // has successfully completed, we can detect this and abort the job, as rolling back a
- // result stage is not possible.
- val stagesToRollback = collectSucceedingStages(sms)
- abortStageWithInvalidRollBack(stagesToRollback)
- // stages which cannot be rolled back were aborted which leads to removing the
- // the dependant job(s) from the active jobs set
- val numActiveJobsWithStageAfterRollback =
- activeJobs.count(job => stagesToRollback.contains(job.finalStage))
- if (numActiveJobsWithStageAfterRollback == 0) {
- logInfo(log"All jobs depending on the indeterminate stage " +
- log"(${MDC(STAGE_ID, stage.id)}) were aborted so this stage is not needed anymore.")
- return
- }
+ if (!sms.shuffleDep.checksumMismatchFullRetryEnabled && stage.isIndeterminate) {
Review Comment:
Will check.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org