You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "attilapiros (via GitHub)" <gi...@apache.org> on 2025/04/17 23:44:01 UTC
[PR] [SPARK-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
attilapiros opened a new pull request, #50630:
URL: https://github.com/apache/spark/pull/50630
### What changes were proposed in this pull request?
This PR aborts the indeterminate partially completed result stage instead of resubmitting it.
### Why are the changes needed?
A result stage compared to shuffle map stage has more output and more state:
- It can use a `FileOutputCommitter` where each task does a Hadoop task commit. In case of a re-submit this will lead to re-commit that Hadoop task (possibly with different content)
- In case of JDBC write it can already inserted all rows of a partitions into the target schema.
As long as rollback of a result stage is not supported (https://issues.apache.org/jira/browse/SPARK-25342) the best we can is abort the stage.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
New unit tests are created for this situation.
### 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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2049762666
##########
core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala:
##########
@@ -56,28 +56,31 @@ class DAGSchedulerEventProcessLoopTester(dagScheduler: DAGScheduler)
dagScheduler.setEventProcessLoop(this)
- private var isProcessing = false
- private val eventQueue = new ListBuffer[DAGSchedulerEvent]()
-
+ private val eventQueue = new LinkedBlockingQueue[DAGSchedulerEvent]()
override def post(event: DAGSchedulerEvent): Unit = {
- if (isProcessing) {
- // `DAGSchedulerEventProcessLoop` is guaranteed to process events sequentially. So we should
- // buffer events for sequent processing later instead of processing them recursively.
- eventQueue += event
- } else {
- try {
- isProcessing = true
- // Forward event to `onReceive` directly to avoid processing event asynchronously.
- onReceive(event)
- } catch {
- case NonFatal(e) => onError(e)
- } finally {
- isProcessing = false
- }
- if (eventQueue.nonEmpty) {
- post(eventQueue.remove(0))
- }
+ // `DAGSchedulerEventProcessLoop` is guaranteed to process events sequentially in the main test
Review Comment:
This part is modified as I have seen the following in the unit-tests.log without it (focus on the thread names: `pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite`, `dag-scheduler-message`):
```
25/04/17 14:15:05.815 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO SparkContext: Created broadcast 2 from broadcast at DAGScheduler.scala:1662
25/04/17 14:15:05.815 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: Submitting 2 missing tasks from ResultStage 2 (DAGSchedulerSuiteRDD 2) (first 15 tasks are for partitions Vector(0, 1))
25/04/17 14:15:05.816 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: Marking ResultStage 2 () as failed due to a fetch failure from ShuffleMapStage 1 (RDD at DAGSchedulerSuite.scala:123)
25/04/17 14:15:05.817 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: ResultStage 2 () failed in 3 ms due to ignored
25/04/17 14:15:05.817 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: Resubmitting ShuffleMapStage 1 (RDD at DAGSchedulerSuite.scala:123) and ResultStage 2 () due to fetch failure
25/04/17 14:15:05.817 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: Executor lost: hostA-exec (epoch 3)
25/04/17 14:15:05.818 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: Shuffle files lost for executor: hostA-exec (epoch 3)
25/04/17 14:15:06.023 dag-scheduler-message INFO DAGSchedulerSuite$MyDAGScheduler: Resubmitting failed stages
25/04/17 14:15:06.024 dag-scheduler-message INFO DAGSchedulerSuite$MyDAGScheduler: Submitting ShuffleMapStage 1 (DAGSchedulerSuiteRDD 0), which has no missing parents
25/04/17 14:15:06.025 dag-scheduler-message INFO MemoryStore: Block broadcast_3 stored as values in memory (estimated size 2.9 KiB, free 2.4 GiB)
25/04/17 14:15:06.025 dag-scheduler-message INFO MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 1825.0 B, free 2.4 GiB)
25/04/17 14:15:06.026 dag-scheduler-message INFO SparkContext: Created broadcast 3 from broadcast at DAGScheduler.scala:1662
25/04/17 14:15:06.027 dag-scheduler-message INFO DAGSchedulerSuite$MyDAGScheduler: Submitting 1 missing tasks from ShuffleMapStage 1 (DAGSchedulerSuiteRDD 0) (first 15 tasks are for partitions Vector(0))
25/04/17 14:15:06.028 dag-scheduler-message INFO DAGSchedulerSuite$MyDAGScheduler: Submitting ShuffleMapStage 0 (DAGSchedulerSuiteRDD 1), which has no missing parents
25/04/17 14:15:06.029 dag-scheduler-message INFO MemoryStore: Block broadcast_4 stored as values in memory (estimated size 2.9 KiB, free 2.4 GiB)
25/04/17 14:15:06.029 dag-scheduler-message INFO MemoryStore: Block broadcast_4_piece0 stored as bytes in memory (estimated size 1826.0 B, free 2.4 GiB)
25/04/17 14:15:06.030 dag-scheduler-message INFO SparkContext: Created broadcast 4 from broadcast at DAGScheduler.scala:1662
25/04/17 14:15:06.030 dag-scheduler-message INFO DAGSchedulerSuite$MyDAGScheduler: Submitting 2 missing tasks from ShuffleMapStage 0 (DAGSchedulerSuiteRDD 1) (first 15 tasks are for partitions Vector(0, 1))
25/04/17 14:15:06.227 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: ShuffleMapStage 0 (RDD at DAGSchedulerSuite.scala:123) finished in 198 ms
25/04/17 14:15:06.228 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: looking for newly runnable stages
25/04/17 14:15:06.228 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: running: HashSet(ShuffleMapStage 1)
25/04/17 14:15:06.229 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: waiting: HashSet(ResultStage 2)
25/04/17 14:15:06.229 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: failed: HashSet()
25/04/17 14:15:06.233 pool-1-thread-1-ScalaTest-running-DAGSchedulerSuite INFO DAGSchedulerSuite$MyDAGScheduler: ShuffleMapStage 1 (RDD at DAGSchedulerSuite.scala:123) finished in 209 ms
```
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093658439
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
@cloud-fan The rollback is not executed when the `FetchFailed` is handled. It was (is) just checking whether a rollback would be needed on that chain of stages (including the result stage) and if it was needed it is checked whether it is possible. If rollback is needed but not possible (as for a `ResultStage` we have any autpot generated) but we do the abort.
See
https://github.com/apache/spark/blob/02f196a4f2c0615826e78fa0f6d99b49b66b81f1/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2177-L2182
So there is no double rollback and abort is protected from double abort:
https://github.com/apache/spark/blob/02f196a4f2c0615826e78fa0f6d99b49b66b81f1/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2883-L2886
The advantages to do this check in `submitMissingTasks`:
- we can handle the case when the executor loss introducing the recalculation of an indeterministic stage
- we are latter in time as we were at the `FetchFailed` so we can make better decision. What I mean if any task finished for the result stage after the `FetchFailed` (as there is race among the events) we can abort the stage and not continuing with partial result coming as a result of the previous run.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2069716779
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2138,7 +2143,8 @@ 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) {
- val rollingBackStages = abortIndeterminateStageChildren(mapStage)._2
+ val stagesToRollback = collectSucceedingStages(mapStage)
+ val rollingBackStages = rollBackStages(stagesToRollback)
Review Comment:
I have refactored `abortIndeterminateStageChildren` into two separate methods:
- collectSucceedingStages
- rollBackStages
This way we do not need returning a tuple anymore and we are not calculating the "All jobs with map stage have been aborted" when it is not needed.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2889709170
thanks, merging to master/4.0/3.5!
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093679044
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
Actually map stages was rolled back by `submitMissingTasks` with the call
```
mapOutputTracker.unregisterAllMapAndMergeOutput(sms.shuffleDep.shuffleId)
```
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2814214867
cc @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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2814730525
> But that's also bad for an indeterminate stage as the data is inconsistent. I mean the committed partitions are coming from a previous old computation and not from the latest one but the resubmitted ones are coming from the new one.
If parent map stage was indeterminate - existing spark code would have already aborted the stage.
As you have pointed out in the test in this PR, there is a gap in the existing impl - which is that when there is a shuffle loss due to executor/host failure (and not due to fetch failure) - the check for determinism was not being performed; and so if shuffle files are lost for an indeterminate stage, the check to abort its child stages was not being done.
This is indeed a bug, which needs to be addressed - and I have proposed two options for it.
But that does not require failing the result stage - even if it is indeterminate.
> So if we write the df to a table and some but not all tasks was successful and a resubmit happened we might have inconsistent result where sum(cnt) won't be 1000 when we load back the data as the resubmit might run on the shuffle map stage which regenerated the random values but with a different distribution of the value from 0 to 10. The complete shuffle map stage re-executed but the result stage not.
This will not happen - please see above.
"some but not all tasks was successful and a resubmit happened" -> if it results in reexecution of the parent stage, job will be aborted.
If it does not result in re-execution of parent stage - the computation is deterministic - it is essentially: "WITH foo AS (SELECT key, count(key) as cnt FROM <constant table> GROUP BY rand) SELECT SUM(cnt) FROM foo" - which will always give same result.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093679044
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
In a sense map stages was actually rolled back by `submitMissingTasks` with the call
```
mapOutputTracker.unregisterAllMapAndMergeOutput(sms.shuffleDep.shuffleId)
```
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2863630749
cc @cloud-fan do you any objections? If not we would like to merge this early next week.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2826383318
> @mridulm regarding option 2 why a return is enough here (and not an abortStage):
It should result in same behavior (all jobs, this stage was part of, have been aborted in that scenario - and we have not added the stage to the runningStages yes).
Having said that, it is indeed better to explictly abort it - even if it does not do much right now : so that we are more robust to code evolution in future.
> Why we need to check whether all jobs should be aborted and not only just one, here:
Stage can be part of multiple concurrent jobs, and not all of them might be getting aborted: some of them might not have started a result stage, and so recoverable.
Only if all of them have been aborted, we can abort the stage.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2087350445
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
+ if (sms.getNextAttemptId > 0) {
+ val stagesToRollback = collectSucceedingStages(sms)
+ rollBackStages(stagesToRollback)
+ // stages which cannot be rolled back were aborted which leads to removing the
+ // the dependant job(s) from the active jobs set
Review Comment:
Why don't we need to do same for the FetchFailed case?
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2094104754
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2129,60 +2149,8 @@ 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) {
- // It's a little tricky to find all the succeeding stages of `mapStage`, because
- // each stage only know its parents not children. Here we traverse the stages from
- // the leaf nodes (the result stages of active jobs), and rollback all the stages
- // in the stage chains that connect to the `mapStage`. To speed up the stage
- // traversing, we collect the stages to rollback first. If a stage needs to
- // rollback, all its succeeding stages need to rollback to.
- val stagesToRollback = HashSet[Stage](mapStage)
-
- def collectStagesToRollback(stageChain: List[Stage]): Unit = {
- if (stagesToRollback.contains(stageChain.head)) {
- stageChain.drop(1).foreach(s => stagesToRollback += s)
- } else {
- stageChain.head.parents.foreach { s =>
- collectStagesToRollback(s :: stageChain)
- }
- }
- }
-
- def generateErrorMessage(stage: Stage): String = {
- "A shuffle map stage with indeterminate output was failed and retried. " +
- s"However, Spark cannot rollback the $stage to re-process the input data, " +
- "and has to fail this job. Please eliminate the indeterminacy by " +
- "checkpointing the RDD before repartition and try again."
- }
-
- activeJobs.foreach(job => collectStagesToRollback(job.finalStage :: Nil))
-
- // The stages will be rolled back after checking
- val rollingBackStages = HashSet[Stage](mapStage)
- stagesToRollback.foreach {
- case mapStage: ShuffleMapStage =>
- val numMissingPartitions = mapStage.findMissingPartitions().length
- if (numMissingPartitions < mapStage.numTasks) {
- if (sc.conf.get(config.SHUFFLE_USE_OLD_FETCH_PROTOCOL)) {
- val reason = "A shuffle map stage with indeterminate output was failed " +
- "and retried. However, Spark can only do this while using the new " +
- "shuffle block fetching protocol. Please check the config " +
- "'spark.shuffle.useOldFetchProtocol', see more detail in " +
- "SPARK-27665 and SPARK-25341."
- abortStage(mapStage, reason, None)
- } else {
- rollingBackStages += mapStage
- }
- }
-
- case resultStage: ResultStage if resultStage.activeJob.isDefined =>
- val numMissingPartitions = resultStage.findMissingPartitions().length
- if (numMissingPartitions < resultStage.numTasks) {
- // TODO: support to rollback result tasks.
- abortStage(resultStage, generateErrorMessage(resultStage), None)
- }
-
- case _ =>
- }
+ val stagesToRollback = collectSucceedingStages(mapStage)
+ val rollingBackStages = validateStageRollBacks(stagesToRollback)
Review Comment:
Yes, this is the fail fast and fail early branch.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093661921
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
+ if (sms.getNextAttemptId > 0) {
+ val stagesToRollback = collectSucceedingStages(sms)
+ rollBackStages(stagesToRollback)
+ // stages which cannot be rolled back were aborted which leads to removing the
+ // the dependant job(s) from the active jobs set
Review Comment:
Here the situation is a bit different as we are in the middle of a `submitMissingTasks`. So we just end this method early as the resubmit of the missing tasks of the stage does not needed when the whole job is aborted.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2814650652
@mridulm
> Only unsuccessful (and so uncommitted) tasks are candidates for (re)execution (and so commit) - not completed tasks.
> So if a partition has completed task commit, it wont be reexecuted - spark ensures this w.r.t use of FileOutputCommitter
But that's also bad for an indeterminate stage as the data is inconsistent. I mean the committed partitions are coming from a previous old computation and not from the latest one but the resubmitted ones are coming from the new one.
To illustrate it:
```
scala> import org.apache.spark.sql.functions.udf
scala> val myudf = udf(() => { val rnd = new java.util.Random(); rnd.nextInt(10)}).asNondeterministic()
scala> spark.udf.register("myudf", myudf)
scala> val df = sql("SELECT rand, count(rand) as cnt from (SELECT myudf() as rand from explode(sequence(1, 1000))) GROUP BY rand")
scala> df.show
+----+---+
|rand|cnt|
+----+---+
| 1|122|
| 6|110|
| 3|111|
| 5| 85|
| 9| 99|
| 4| 94|
| 8| 93|
| 7| 88|
| 2| 98|
| 0|100|
+----+---+
scala> df.selectExpr("sum(cnt)").show
+--------+
|sum(cnt)|
+--------+
| 1000|
+--------+
```
So if we write the `df` to a table and some but not all tasks was successful and a resubmit happened we might have inconsistent result where `sum(cnt)` won't be 1000 when we load back the data as the resubmit might run on the shuffle map stage which regenerated the random values but with a different distribution of the value from 0 to 10. The complete shuffle map stage re-executed but the result stage not.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093676889
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
I will extend the comments and rename `rollBackStages` to `validateStageRollBacks`.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2087348255
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2346,6 +2308,72 @@ private[spark] class DAGScheduler(
}
}
+ private def collectSucceedingStages(mapStage: ShuffleMapStage): HashSet[Stage] = {
Review Comment:
I assume these methods just moved code from https://github.com/apache/spark/pull/50630/files#diff-85de35b2e85646ed499c545a3be1cd3ffd525a88aae835a9c621f877eebadcb6L2132 ?
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2889712758
@attilapiros can you open a 3.5 backport PR? thanks!
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2825460926
@mridulm regarding option 2 why a return is enough here (and not an `abortStage`):
https://github.com/mridulm/spark/blob/9301803d018b3ba5195359232bf015dea5635759/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1559
meanwhile when there is an exception at task creation an `abortStage` is called explicitly:
https://github.com/mridulm/spark/blob/9301803d018b3ba5195359232bf015dea5635759/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1616
Why we need to check whether all jobs should be aborted and not only just one, here:
https://github.com/mridulm/spark/blob/9301803d018b3ba5195359232bf015dea5635759/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2370
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2848520366
@mridulm
> But then, I am too close to this :-)
Definitely you should be credited here. Is there a way to list you as co-author?
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093981271
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2129,60 +2149,8 @@ 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) {
- // It's a little tricky to find all the succeeding stages of `mapStage`, because
- // each stage only know its parents not children. Here we traverse the stages from
- // the leaf nodes (the result stages of active jobs), and rollback all the stages
- // in the stage chains that connect to the `mapStage`. To speed up the stage
- // traversing, we collect the stages to rollback first. If a stage needs to
- // rollback, all its succeeding stages need to rollback to.
- val stagesToRollback = HashSet[Stage](mapStage)
-
- def collectStagesToRollback(stageChain: List[Stage]): Unit = {
- if (stagesToRollback.contains(stageChain.head)) {
- stageChain.drop(1).foreach(s => stagesToRollback += s)
- } else {
- stageChain.head.parents.foreach { s =>
- collectStagesToRollback(s :: stageChain)
- }
- }
- }
-
- def generateErrorMessage(stage: Stage): String = {
- "A shuffle map stage with indeterminate output was failed and retried. " +
- s"However, Spark cannot rollback the $stage to re-process the input data, " +
- "and has to fail this job. Please eliminate the indeterminacy by " +
- "checkpointing the RDD before repartition and try again."
- }
-
- activeJobs.foreach(job => collectStagesToRollback(job.finalStage :: Nil))
-
- // The stages will be rolled back after checking
- val rollingBackStages = HashSet[Stage](mapStage)
- stagesToRollback.foreach {
- case mapStage: ShuffleMapStage =>
- val numMissingPartitions = mapStage.findMissingPartitions().length
- if (numMissingPartitions < mapStage.numTasks) {
- if (sc.conf.get(config.SHUFFLE_USE_OLD_FETCH_PROTOCOL)) {
- val reason = "A shuffle map stage with indeterminate output was failed " +
- "and retried. However, Spark can only do this while using the new " +
- "shuffle block fetching protocol. Please check the config " +
- "'spark.shuffle.useOldFetchProtocol', see more detail in " +
- "SPARK-27665 and SPARK-25341."
- abortStage(mapStage, reason, None)
- } else {
- rollingBackStages += mapStage
- }
- }
-
- case resultStage: ResultStage if resultStage.activeJob.isDefined =>
- val numMissingPartitions = resultStage.findMissingPartitions().length
- if (numMissingPartitions < resultStage.numTasks) {
- // TODO: support to rollback result tasks.
- abortStage(resultStage, generateErrorMessage(resultStage), None)
- }
-
- case _ =>
- }
+ val stagesToRollback = collectSucceedingStages(mapStage)
+ val rollingBackStages = validateStageRollBacks(stagesToRollback)
Review Comment:
to double check: given we will handle this in `submitMissingTasks`, the code here is not really necessary, but just for better perf to rollback/abort stages earlier?
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2094110070
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2346,6 +2314,72 @@ private[spark] class DAGScheduler(
}
}
+ private def collectSucceedingStages(mapStage: ShuffleMapStage): HashSet[Stage] = {
+ // TODO: perhaps materialize this if we are going to compute it often enough ?
+ // It's a little tricky to find all the succeeding stages of `mapStage`, because
+ // each stage only know its parents not children. Here we traverse the stages from
+ // the leaf nodes (the result stages of active jobs), and rollback all the stages
+ // in the stage chains that connect to the `mapStage`. To speed up the stage
+ // traversing, we collect the stages to rollback first. If a stage needs to
+ // rollback, all its succeeding stages need to rollback to.
+ val succeedingStages = HashSet[Stage](mapStage)
+
+ def collectSucceedingStagesInternal(stageChain: List[Stage]): Unit = {
+ if (succeedingStages.contains(stageChain.head)) {
+ stageChain.drop(1).foreach(s => succeedingStages += s)
+ } else {
+ stageChain.head.parents.foreach { s =>
+ collectSucceedingStagesInternal(s :: stageChain)
+ }
+ }
+ }
+ activeJobs.foreach(job => collectSucceedingStagesInternal(job.finalStage :: Nil))
+ succeedingStages
+ }
+
+ /**
Review Comment:
Ok and I rename this method to `abortInvalidStageRollBacks`
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2087344954
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
```suggestion
// already executed at least once
```
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2094153340
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2346,6 +2314,72 @@ private[spark] class DAGScheduler(
}
}
+ private def collectSucceedingStages(mapStage: ShuffleMapStage): HashSet[Stage] = {
+ // TODO: perhaps materialize this if we are going to compute it often enough ?
+ // It's a little tricky to find all the succeeding stages of `mapStage`, because
+ // each stage only know its parents not children. Here we traverse the stages from
+ // the leaf nodes (the result stages of active jobs), and rollback all the stages
+ // in the stage chains that connect to the `mapStage`. To speed up the stage
+ // traversing, we collect the stages to rollback first. If a stage needs to
+ // rollback, all its succeeding stages need to rollback to.
+ val succeedingStages = HashSet[Stage](mapStage)
+
+ def collectSucceedingStagesInternal(stageChain: List[Stage]): Unit = {
+ if (succeedingStages.contains(stageChain.head)) {
+ stageChain.drop(1).foreach(s => succeedingStages += s)
+ } else {
+ stageChain.head.parents.foreach { s =>
+ collectSucceedingStagesInternal(s :: stageChain)
+ }
+ }
+ }
+ activeJobs.foreach(job => collectSucceedingStagesInternal(job.finalStage :: Nil))
+ succeedingStages
+ }
+
+ /**
Review Comment:
`abortStageWithInvalidRollBack` is even better.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093981415
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2346,6 +2314,72 @@ private[spark] class DAGScheduler(
}
}
+ private def collectSucceedingStages(mapStage: ShuffleMapStage): HashSet[Stage] = {
+ // TODO: perhaps materialize this if we are going to compute it often enough ?
+ // It's a little tricky to find all the succeeding stages of `mapStage`, because
+ // each stage only know its parents not children. Here we traverse the stages from
+ // the leaf nodes (the result stages of active jobs), and rollback all the stages
+ // in the stage chains that connect to the `mapStage`. To speed up the stage
+ // traversing, we collect the stages to rollback first. If a stage needs to
+ // rollback, all its succeeding stages need to rollback to.
+ val succeedingStages = HashSet[Stage](mapStage)
+
+ def collectSucceedingStagesInternal(stageChain: List[Stage]): Unit = {
+ if (succeedingStages.contains(stageChain.head)) {
+ stageChain.drop(1).foreach(s => succeedingStages += s)
+ } else {
+ stageChain.head.parents.foreach { s =>
+ collectSucceedingStagesInternal(s :: stageChain)
+ }
+ }
+ }
+ activeJobs.foreach(job => collectSucceedingStagesInternal(job.finalStage :: Nil))
+ succeedingStages
+ }
+
+ /**
Review Comment:
can we add a bit more comments? This method does not only do validation, but also have side effects to abort the stages and fail their corresponding jobs if the stage can't be rolled back.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2814502857
The "java.lang.OutOfMemoryError: Java heap space" in the pyspark-pandas-connect-part2 is unrelated.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2814578661
> It can use a FileOutputCommitter where each task does a Hadoop task commit. In case of a re-submit this will lead to re-commit that Hadoop task (possibly with different content)
Only unsuccessful (and so uncommitted) tasks are candidates for (re)execution (and so commit) - not completed tasks.
So if a partition has completed task commit, it wont be reexecuted - spark ensures this w.r.t use of `FileOutputCommitter`
> In case of JDBC write it can already inserted all rows of a partitions into the target schema.
As discussed [here](https://github.com/apache/spark/pull/50033#issuecomment-2808115624), this is a bug in jdbc implementation - the txn commit should be done in a task commit, not as part of `foreachPartition(savePartition)`.
It is not expected to work correctly in all scenarios, and in the case observed, it did end up failing.
> But this is not enough as a FetchFailed from a determinate stage can lead to an executor loss and a re-compute of the indeterminate parent of the result stage as shown in the attached unittest.
The fix for this is to handle something similar [to this](https://github.com/apache/spark/blob/8b33a832b2ba4f8bb7ed34dac50778bf8cbcfa13/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2131).
I have sketched a rough impl here for reference (it is just illustrative ! and to convey what I was talking about).
Assuming I am not missing anything, it should help you with this fixing this issue.
* [Option 1](https://github.com/apache/spark/compare/master...mridulm:spark:prototype-for-attila-indeterminism-discussion-option-1) handles indeterminate impact when processing shuffle data loss.
* [Option 2](https://github.com/apache/spark/compare/master...mridulm:spark:prototype-for-attila-indeterminism-discussion-option-2-subMissingTasks) does this when computing an indeterminate stage.
Option 1 is much more aggressive with cleanup, but might spuriously kills jobs a lot more than required.
If option 2 is correct, I would prefer that - as it is much more conservative with abort'ing stages/failing jobs.
(I have adapted the tests you included in this PR for both)
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2843221263
@mridulm I am addig your option2 solution but I would like to do some changes based on it
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2069715044
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2337,10 +2344,8 @@ private[spark] class DAGScheduler(
"checkpointing the RDD before repartition and try again."
}
- activeJobs.foreach(job => collectStagesToRollback(job.finalStage :: Nil))
-
// The stages will be rolled back after checking
- val rollingBackStages = HashSet[Stage](mapStage)
+ val rollingBackStages = HashSet[Stage]()
Review Comment:
The old `mapstage` is part of `stagesToRollback` as the `succeedingStages` in the `collectSucceedingStages` was including the `mapStage`.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093658439
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
@cloud-fan The rollback is not executed when the `FetchFailed` is handled. It was (is) just checking whether a rollback would be needed on that chain of stages (including the result stage) and if it was needed it is checked whether it is possible. If rollback is needed but not possible (as for a `ResultStage` we have any autpot generated) but we do the abort.
See
https://github.com/apache/spark/blob/02f196a4f2c0615826e78fa0f6d99b49b66b81f1/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2177-L2182
So there is no double rollback and abort is protected from double abort:
https://github.com/apache/spark/blob/02f196a4f2c0615826e78fa0f6d99b49b66b81f1/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2883-L2886
The advantages to do this check in `submitMissingTasks`:
- we can handle the case when the executor loss introducing the recalculation of an indeterministic stage
- we are latter in time as we were at the `FetchFailed` so we can make better decision. What I mean if any task finished for the result stage successfully after the `FetchFailed` (as there is race among the events) we can abort the stage and not continuing with partial result coming as a result of the previous run.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #50630: [SPARK-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages
URL: https://github.com/apache/spark/pull/50630
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2069585506
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2362,12 +2363,15 @@ private[spark] class DAGScheduler(
if (numMissingPartitions < resultStage.numTasks) {
// TODO: support to rollback result tasks.
abortStage(resultStage, generateErrorMessage(resultStage), None)
- numAbortedJobs += 1
}
case _ =>
}
- (numAbortedJobs >= numJobsWithStage, rollingBackStages)
+
+ val numActiveJobsWithStageAfterRollback =
+ activeJobs.count(job => stagesToRollback.contains(job.finalStage))
Review Comment:
@mridulm I made this change as when the shuffle map stage was aborted because of SHUFFLE_USE_OLD_FETCH_PROTOCOL that should be counted into `numAborted` too.
And it is easier to use the `activeJobs` again as it is updated at:
https://github.com/apache/spark/blob/3028c56467352924e3c44554b28b508355616e8b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L907
Called via the call chain of `cleanupStateForJobAndIndependentStages()` <- `failJobAndIndependentStages()` <- `abortStage()`.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2843865879
> I am addig your option2 solution but I would like to do some changes based on it
Please do ! It was just a rough sketch :-)
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2072322347
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2337,10 +2344,8 @@ private[spark] class DAGScheduler(
"checkpointing the RDD before repartition and try again."
}
- activeJobs.foreach(job => collectStagesToRollback(job.finalStage :: Nil))
-
// The stages will be rolled back after checking
- val rollingBackStages = HashSet[Stage](mapStage)
+ val rollingBackStages = HashSet[Stage]()
Review Comment:
We will end up calling `findMissingPartitions` for the mapStage, when we know it will have missing partitions.
That would be an improvement to consider.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2848670694
@mridulm yes you are right still I would like to give you the credit too if there is a way
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2814560313
After the test was restarted the error is resolved.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2072347515
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2337,10 +2344,8 @@ private[spark] class DAGScheduler(
"checkpointing the RDD before repartition and try again."
}
- activeJobs.foreach(job => collectStagesToRollback(job.finalStage :: Nil))
-
// The stages will be rolled back after checking
- val rollingBackStages = HashSet[Stage](mapStage)
+ val rollingBackStages = HashSet[Stage]()
Review Comment:
This was the case even before this PR, is not it?
But this makes me wonder why we materialize the missing partitions containing the exact missing partitions itself?
We already have a `numAvailableOutputs` which is based on a counter in `MapOutputTracker`: https://github.com/apache/spark/blob/db59634edaf8e1dc587077edd5a2bc7955b3f357/core/src/main/scala/org/apache/spark/MapOutputTracker.scala#L143-L148
When `numAvailableOutputs` is nonzero we need to roll it back.
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2072349781
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2337,10 +2344,8 @@ private[spark] class DAGScheduler(
"checkpointing the RDD before repartition and try again."
}
- activeJobs.foreach(job => collectStagesToRollback(job.finalStage :: Nil))
-
// The stages will be rolled back after checking
- val rollingBackStages = HashSet[Stage](mapStage)
+ val rollingBackStages = HashSet[Stage]()
Review Comment:
https://github.com/apache/spark/pull/50630/commits/dd141c585595b4e24d8b9ad43ef32df0c67d9110
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2087349665
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
Can we say more in the comments? So this case is for when the shuffle stage was not failed due to FetchFailed, we also need to roll back the result stage?
BTW will it cause repeated result stage rollback? e.g. we roll back it once when hitting FetchFailed, and roll back it again here then the failed shuffle stage is resubmitted.
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093706546
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -1552,6 +1552,20 @@ private[spark] class DAGScheduler(
// `findMissingPartitions()` returns all partitions every time.
stage match {
case sms: ShuffleMapStage if stage.isIndeterminate && !sms.isAvailable =>
+ // already executed atleast once
Review Comment:
Addressed in https://github.com/apache/spark/pull/50630/commits/ec7f3738efe17d87850f150cbeb0e69418873995
--
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-51272][CORE] Aborting instead of continuing partially completed indeterminate result stage at ResubmitFailedStages [spark]
Posted by "attilapiros (via GitHub)" <gi...@apache.org>.
attilapiros commented on code in PR #50630:
URL: https://github.com/apache/spark/pull/50630#discussion_r2093659046
##########
core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala:
##########
@@ -2346,6 +2308,72 @@ private[spark] class DAGScheduler(
}
}
+ private def collectSucceedingStages(mapStage: ShuffleMapStage): HashSet[Stage] = {
Review Comment:
Yes, you are right!
--
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-51272][CORE] Aborting instead of re-submitting of partially completed indeterminate result stage [spark]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on PR #50630:
URL: https://github.com/apache/spark/pull/50630#issuecomment-2848642030
> Definitely you should be credited here. Is there a way to list you as co-author?
I meant in turns of objectivity !
If there is a blind spot, both of us might not be seeing it
--
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