Hi Mridul, thanks for the explanation, it's clear to me now, Thanks! Mridul Muralidharan <mri...@gmail.com> 于2023年10月20日周五 11:15写道:
> To add my response - what I described (w.r.t failing job) applies only to > ResultStage. > It walks the lineage DAG to identify all indeterminate parents to rollback. > If there are only ShuffleMapStages in the set of stages to rollback, it > will simply discard their output, rollback all of them, and then retry > these stages (same shuffle-id, a new stage attempt) > > > Regards, > Mridul > > > > On Thu, Oct 19, 2023 at 10:08 PM Mridul Muralidharan <mri...@gmail.com> > wrote: > > > > > Good question, and ResultStage is actually special cased in spark as its > > output could have already been consumed (for example collect() to driver, > > etc) - and so if it is one of the stages which needs to be rolled back, > the > > job is aborted. > > > > To illustrate, see the following: > > -- snip -- > > > > package org.apache.spark > > > > > > import scala.reflect.ClassTag > > > > import org.apache.spark._ > > import org.apache.spark.rdd.{DeterministicLevel, RDD} > > > > class DelegatingRDD[E: ClassTag](delegate: RDD[E]) extends > RDD[E](delegate) { > > > > override def compute(split: Partition, context: TaskContext): > Iterator[E] = { > > delegate.compute(split, context) > > } > > > > override protected def getPartitions: Array[Partition] = > > delegate.partitions > > } > > > > class IndeterminateRDD[E: ClassTag](delegate: RDD[E]) extends > DelegatingRDD[E](delegate) { > > override def getOutputDeterministicLevel: DeterministicLevel.Value = > DeterministicLevel.INDETERMINATE > > } > > > > class FailingRDD[E: ClassTag](delegate: RDD[E]) extends > DelegatingRDD[E](delegate) { > > override def compute(split: Partition, context: TaskContext): > Iterator[E] = { > > val tc = TaskContext.get > > if (tc.stageAttemptNumber() == 0 && tc.partitionId() == 0 && > tc.attemptNumber() == 0) { > > // Wait for all tasks to be done, then call exit > > Thread.sleep(5000) > > System.exit(-1) > > } > > delegate.compute(split, context) > > } > > } > > > > // Make sure test_output directory is deleted before running this. > > // > > object Test { > > > > def main(args: Array[String]): Unit = { > > val conf = new SparkConf().setMaster("local-cluster[4,1,1024]") > > val sc = new SparkContext(conf) > > > > val mapperRdd = new IndeterminateRDD(sc.parallelize(0 until 10000, > 20).map(v => (v, v))) > > val resultRdd = new FailingRDD(mapperRdd.groupByKey()) > > resultRdd.saveAsTextFile("test_output") > > } > > } > > > > -- snip -- > > > > > > > > Here, the mapper stage has been forced to be INDETERMINATE. > > In the reducer stage, the first attempt to compute partition 0 will wait > for a bit and then exit - since the master is a local-cluster, this results > in FetchFailure when the second attempt of partition 0 tries to fetch > shuffle data. > > When spark tries to regenerate parent shuffle output, it sees that the > parent is INDETERMINATE - and so fails the entire job.with the message: > > " > > org.apache.spark.SparkException: Job aborted due to stage failure: A > shuffle map stage with indeterminate output was failed and retried. > However, Spark cannot rollback the ResultStage 1 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. > > " > > > > This is coming from here < > https://github.com/apache/spark/blob/28292d51e7dbe2f3488e82435abb48d3d31f6044/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L2090> > - when rolling back stages, if spark determines that a ResultStage needs to > be rolled back due to loss of INDETERMINATE output, it will fail the job. > > > > Hope this clarifies. > > Regards, > > Mridul > > > > > > On Thu, Oct 19, 2023 at 10:04 AM Keyong Zhou <zho...@apache.org> wrote: > > > >> In fact, I'm wondering if Spark will rerun the whole reduce > >> ShuffleMapStage > >> if its upstream ShuffleMapStage is INDETERMINATE and rerun. > >> > >> Keyong Zhou <zho...@apache.org> 于2023年10月19日周四 23:00写道: > >> > >> > Thanks Erik for bringing up this question, I'm also curious about the > >> > answer, any feedback is appreciated. > >> > > >> > Thanks, > >> > Keyong Zhou > >> > > >> > Erik fang <fme...@gmail.com> 于2023年10月19日周四 22:16写道: > >> > > >> >> Mridul, > >> >> > >> >> sure, I totally agree SPARK-25299 is a much better solution, as long > >> as we > >> >> can get it from spark community > >> >> (btw, private[spark] of RDD.outputDeterministicLevel is no big deal, > >> >> celeborn already has spark-integration code with [spark] scope) > >> >> > >> >> I also have a question about INDETERMINATE stage recompute, and may > >> need > >> >> your help > >> >> The rule for INDETERMINATE ShuffleMapStage rerun is reasonable, > >> however, I > >> >> don't find related logic for INDETERMINATE ResultStage rerun in > >> >> DAGScheduler > >> >> If INDETERMINATE ShuffleMapStage got entirely recomputed, the > >> >> corresponding ResultStage should be entirely recomputed as well, per > my > >> >> understanding > >> >> > >> >> I found https://issues.apache.org/jira/browse/SPARK-25342 to > rollback > >> a > >> >> ResultStage but it was not merged > >> >> Do you know any context or related ticket for INDETERMINATE > ResultStage > >> >> rerun? > >> >> > >> >> Thanks in advance! > >> >> > >> >> Regards, > >> >> Erik > >> >> > >> >> On Tue, Oct 17, 2023 at 4:23 AM Mridul Muralidharan < > mri...@gmail.com> > >> >> wrote: > >> >> > >> >> > > >> >> > > >> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fme...@gmail.com> > wrote: > >> >> > > >> >> >> Hi Mridul, > >> >> >> > >> >> >> For a), > >> >> >> DagScheduler uses Stage.isIndeterminate() and RDD.isBarrier() > >> >> >> < > >> >> > >> > https://github.com/apache/spark/blob/3e2470de7ea8b97dcdd8875ef25f044998fb7588/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1975 > >> >> > > >> >> >> to decide whether the whole stage needs to be recomputed > >> >> >> I think we can pass the same information to Celeborn in > >> >> >> ShuffleManager.registerShuffle() > >> >> >> < > >> >> > >> > https://github.com/apache/spark/blob/721ea9bbb2ff77b6d2f575fdca0aeda84990cc3b/core/src/main/scala/org/apache/spark/shuffle/ShuffleManager.scala#L39 > >> >, > >> >> since > >> >> >> RDD in ShuffleDependency contains the RDD object > >> >> >> It seems Stage.isIndeterminate() is unreadable from > >> ShuffleDependency, > >> >> >> but luckily rdd is used internally > >> >> >> > >> >> >> def isIndeterminate: Boolean = { > >> >> >> rdd.outputDeterministicLevel == DeterministicLevel.INDETERMINATE > >> >> >> } > >> >> >> > >> >> >> Relies on internal implementation is not good, but doable. > >> >> >> I don't expect spark RDD/Stage implementation changes frequently, > >> and > >> >> we > >> >> >> can discuss with Spark community for a RDD isIndeterminate API if > >> they > >> >> >> change it in the future > >> >> >> > >> >> > > >> >> > > >> >> > Only RDD.getOutputDeterministicLevel is publicly exposed, > >> >> > RDD.outputDeterministicLevel is not and it is private[spark]. > >> >> > While I dont expect changes to this, it is inherently unstable to > >> depend > >> >> > on it. > >> >> > > >> >> > Btw, please see the discussion with Sungwoo Park, if Celeborn is > >> >> > maintaining a reducer oriented view, you will need to recompute all > >> the > >> >> > mappers anyway - what you might save is the subset of reducer > >> partitions > >> >> > which can be skipped if it is DETERMINATE. > >> >> > > >> >> > > >> >> > > >> >> > > >> >> >> > >> >> >> for c) > >> >> >> I also considered a similar solution in celeborn > >> >> >> Celeborn (LifecycleManager) can get the full picture of remaining > >> >> shuffle > >> >> >> data from previous stage attempt and reuse it in stage recompute > >> >> >> , and the whole process will be transparent to Spark/DagScheduler > >> >> >> > >> >> > > >> >> > Celeborn does not have visibility into this - and this is > potentially > >> >> > subject to invasive changes in Apache Spark as it evolves. > >> >> > For example, I recently merged a couple of changes which would make > >> this > >> >> > different in master compared to previous versions. > >> >> > Until the remote shuffle service SPIP is implemented and these are > >> >> > abstracted out & made pluggable, it will continue to be quite > >> volatile. > >> >> > > >> >> > Note that the behavior for 3.5 and older is known - since Spark > >> versions > >> >> > have been released - it is the behavior in master and future > >> versions of > >> >> > Spark which is subject to change. > >> >> > So delivering on SPARK-25299 would future proof all remote shuffle > >> >> > implementations. > >> >> > > >> >> > > >> >> > Regards, > >> >> > Mridul > >> >> > > >> >> > > >> >> > > >> >> >> > >> >> >> Per my perspective, leveraging partial stage recompute and > >> >> >> remaining shuffle data needs a lot of work to do in Celeborn > >> >> >> I prefer to implement a simple whole stage recompute first with > >> >> interface > >> >> >> defined with recomputeAll = true flag, and explore partial stage > >> >> recompute > >> >> >> in seperate ticket as future optimization > >> >> >> How do you think about it? > >> >> >> > >> >> >> Regards, > >> >> >> Erik > >> >> >> > >> >> >> > >> >> >> On Sat, Oct 14, 2023 at 4:50 PM Mridul Muralidharan < > >> mri...@gmail.com> > >> >> >> wrote: > >> >> >> > >> >> >>> > >> >> >>> > >> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan < > >> mri...@gmail.com > >> >> > > >> >> >>> wrote: > >> >> >>> > >> >> >>>> > >> >> >>>> A reducer oriented view of shuffle, especially without > >> replication, > >> >> >>>> could indeed be susceptible to this issue you described (a > single > >> >> fetch > >> >> >>>> failure would require all mappers to need to be recomputed) - > >> note, > >> >> not > >> >> >>>> necessarily all reducers to be recomputed though. > >> >> >>>> > >> >> >>>> Note that I have not looked much into Celeborn specifically on > >> this > >> >> >>>> aspect yet, so my comments are *fairly* centric to Spark > internals > >> >> :-) > >> >> >>>> > >> >> >>>> Regards, > >> >> >>>> Mridul > >> >> >>>> > >> >> >>>> > >> >> >>>> On Sat, Oct 14, 2023 at 3:36 AM Sungwoo Park <glap...@gmail.com > > > >> >> wrote: > >> >> >>>> > >> >> >>>>> Hello, > >> >> >>>>> > >> >> >>>>> (Sorry for sending the same message again.) > >> >> >>>>> > >> >> >>>>> From my understanding, the current implementation of Celeborn > >> makes > >> >> it > >> >> >>>>> hard to find out which mapper should be re-executed when a > >> >> partition cannot > >> >> >>>>> be read, and we should re-execute all the mappers in the > upstream > >> >> stage. If > >> >> >>>>> we can find out which mapper/partition should be re-executed, > the > >> >> current > >> >> >>>>> logic of stage recomputation could be (partially or totally) > >> reused. > >> >> >>>>> > >> >> >>>>> Regards, > >> >> >>>>> > >> >> >>>>> --- Sungwoo > >> >> >>>>> > >> >> >>>>> On Sat, Oct 14, 2023 at 5:24 PM Mridul Muralidharan < > >> >> mri...@gmail.com> > >> >> >>>>> wrote: > >> >> >>>>> > >> >> >>>>>> > >> >> >>>>>> Hi, > >> >> >>>>>> > >> >> >>>>>> Spark will try to minimize the recomputation cost as much as > >> >> >>>>>> possible. > >> >> >>>>>> For example, if parent stage was DETERMINATE, it simply needs > to > >> >> >>>>>> recompute the missing (mapper) partitions (which resulted in > >> fetch > >> >> >>>>>> failure). Note, this by itself could require further > >> recomputation > >> >> in the > >> >> >>>>>> DAG if the inputs required to comput the parent partitions are > >> >> missing, and > >> >> >>>>>> so on - so it is dynamic. > >> >> >>>>>> > >> >> >>>>>> Regards, > >> >> >>>>>> Mridul > >> >> >>>>>> > >> >> >>>>>> On Sat, Oct 14, 2023 at 2:30 AM Sungwoo Park < > >> >> o...@pl.postech.ac.kr> > >> >> >>>>>> wrote: > >> >> >>>>>> > >> >> >>>>>>> > a) If one or more tasks for a stage (and so its shuffle id) > >> is > >> >> >>>>>>> going to be > >> >> >>>>>>> > recomputed, if it is an INDETERMINATE stage, all shuffle > >> output > >> >> >>>>>>> will be > >> >> >>>>>>> > discarded and it will be entirely recomputed (see here > >> >> >>>>>>> > < > >> >> >>>>>>> > >> >> > >> > https://github.com/apache/spark/blob/3e2470de7ea8b97dcdd8875ef25f044998fb7588/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1477 > >> >> >>>>>>> > > >> >> >>>>>>> > ). > >> >> >>>>>>> > >> >> >>>>>>> If a reducer (in a downstream stage) fails to read data, can > we > >> >> find > >> >> >>>>>>> out > >> >> >>>>>>> which tasks should recompute their output? From the previous > >> >> >>>>>>> discussion, I > >> >> >>>>>>> thought this was hard (in the current implementation), and we > >> >> should > >> >> >>>>>>> re-execute all tasks in the upstream stage. > >> >> >>>>>>> > >> >> >>>>>>> Thanks, > >> >> >>>>>>> > >> >> >>>>>>> --- Sungwoo > >> >> >>>>>>> > >> >> >>>>>> > >> >> > >> > > >> > > >