Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Sungwoo Park <or...@pl.postech.ac.kr>]

Hi Keyong,

Instead of picking up a new shuffleId, can we reuse an existing shuffleId 
after unregistering it? If the following plan worked, it would further 
simplify the implementation:

1. Downstream tasks fail because of read failures.
2. All active downstream tasks are killed, so the shuffleId is not used.
3. An upstream vertex unregisters the shuffleId.
4. The upstream vertex is re-executed normally. This re-execution 
automatically registers the same shuffleId.

In summary, we would like to go back in time before the upstream vertex 
started by cleaning up the shuffleId. Could you please give a comment on 
this plan?

Thank you,

--- Sungwoo

On Sat, 30 Sep 2023, Keyong Zhou wrote:

> Hi Sungwoo,
>
> I think your approach works with current architecture of Celeborn,
> and interpreting IOException when reading as read failure makes
> sense. Currently only when CommitFiles fails will LifecycleManager
> announce data lost.
>
> Thanks,
> Keyong Zhou
>
> Sungwoo Park <or...@pl.postech.ac.kr> 于2023年9月29日周五 22:05?道：
>
>>> Since the partition split has a good chance to contain data from almost
>> all
>>> upstream
>>> mapper tasks, the cost of re-computing all upstream tasks may have little
>>> difference
>>> to re-computing the actual mapper tasks in most cases. Of course it's not
>>> always true.
>>>
>>> To change from 'complete' to 'incomplete' also needs to refactor Worker's
>>> logic, which
>>> currently assumes that the succeeded attempts will not be changed after
>>> final committing
>>> files.
>>>
>>>> a subset of succeeded attempts. In Erik's proposal, the whole upstream
>>>> stage will be rerun when data lost.
>>
>> Thank you for your response --- things are now much clearer.
>>
>> From your comments shown above, let me assume that:
>>
>> 1. The whole upstream stage is rerun in the case of read failure.
>>
>> 2. Currently it is not easy to change the state of a shuffleId from
>> 'complete' to 'incomplete'.
>>
>> Then, for Celeborn-MR3, I would like to experiment with the following
>> approach:
>>
>> 1. If read failures occur for shuffleId #1, we pick up a new shuffleId
>> #2.
>>
>> 2. The upstream stage (or Vertex in the case of MR3) re-executes all tasks
>> again, but writes the output to shuffleId #2.
>>
>> 3. Tasks in the downstream stage re-try by reading from shuffleId #2.
>>
>> Do you think this approach makes sense under the current architecture of
>> Celeborn? If this approach is feasible, MR3 only needs to be notified of
>> read failures due to lost data by Celeborn ShuffleClient. Or, we could
>> just interpret IOException from Celeborn ShuffleClient as read failures,
>> in which case we can implement stage recompute without requiring any
>> extension of Celeborn. (However, it would be great if Celeborn
>> ShuffleClient could announce lost data explicitly.)
>>
>> An industrial user of Hive-MR3-Celeborn is trying hard to save disk usage
>> on Celeborn workers (which use SSDs of limited capacity), so stage
>> recompute would be a great new feature to them.
>>
>> Thank you,
>>
>> --- Sungwoo
>>
>>
>>
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Erik fang <fm...@gmail.com>]

Hi Mridul, Your explanation is clear and great

Thank you so much!

On Fri, Oct 20, 2023 at 11:59 AM Keyong Zhou <zh...@apache.org> wrote:

> Hi Mridul, thanks for the explanation, it's clear to me now, Thanks!
>
> Mridul Muralidharan <mr...@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 <mr...@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 <zh...@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 <zh...@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 <fm...@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 <
> > mridul@gmail.com>
> > >> >> wrote:
> > >> >>
> > >> >> >
> > >> >> >
> > >> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <
> > >> mridul@gmail.com>
> > >> >> >> wrote:
> > >> >> >>
> > >> >> >>>
> > >> >> >>>
> > >> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <
> > >> mridul@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 <
> glapark@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 <
> > >> >> mridul@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 <
> > >> >> orpl@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
> > >> >> >>>>>>>
> > >> >> >>>>>>
> > >> >>
> > >> >
> > >>
> > >
> >
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Keyong Zhou <zh...@apache.org>]

Hi Mridul, thanks for the explanation, it's clear to me now, Thanks!

Mridul Muralidharan <mr...@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 <mr...@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 <zh...@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 <zh...@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 <fm...@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 <
> mridul@gmail.com>
> >> >> wrote:
> >> >>
> >> >> >
> >> >> >
> >> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <
> >> mridul@gmail.com>
> >> >> >> wrote:
> >> >> >>
> >> >> >>>
> >> >> >>>
> >> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <
> >> mridul@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 <glapark@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 <
> >> >> mridul@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 <
> >> >> orpl@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
> >> >> >>>>>>>
> >> >> >>>>>>
> >> >>
> >> >
> >>
> >
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

Yes, DAGScheduler is dealing with it at a stage level - and so individual
RDD’s DeterministicLevel  would be handled in order to determine the
stage’s level.

Regards,
Mridul


On Fri, Nov 3, 2023 at 9:45 AM Keyong Zhou <zh...@apache.org> wrote:

> I checked RDD#getOutputDeterministicLevel and find that if an RDD's
> upstream is INDETERMINATE,
> then it's also INDETERMINATE.
>
> Thanks,
> Keyong Zhou
>
> Keyong Zhou <zh...@apache.org> 于2023年11月3日周五 19:57写道：
>
> > Hi Mridul,
> >
> > I still have a question. DAGScheduler#submitMissingTasks will
> > only unregisterAllMapAndMergeOutput
> > if the current ShuffleMapStage is Indeterminate. What if the current
> stage
> > is determinate, but its
> > upstream stage is Indeterminate, and its upstream stage is rerun?
> >
> > Thanks,
> > Keyong Zhou
> >
> > Mridul Muralidharan <mr...@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 <mr...@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 <zh...@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 <zh...@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 <fm...@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 <
> >> mridul@gmail.com>
> >> >> >> wrote:
> >> >> >>
> >> >> >> >
> >> >> >> >
> >> >> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <
> >> >> mridul@gmail.com>
> >> >> >> >> wrote:
> >> >> >> >>
> >> >> >> >>>
> >> >> >> >>>
> >> >> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <
> >> >> mridul@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 <
> >> glapark@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 <
> >> >> >> mridul@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 <
> >> >> >> orpl@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
> >> >> >> >>>>>>>
> >> >> >> >>>>>>
> >> >> >>
> >> >> >
> >> >>
> >> >
> >>
> >
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Keyong Zhou <zh...@apache.org>]

I checked RDD#getOutputDeterministicLevel and find that if an RDD's
upstream is INDETERMINATE,
then it's also INDETERMINATE.

Thanks,
Keyong Zhou

Keyong Zhou <zh...@apache.org> 于2023年11月3日周五 19:57写道：

> Hi Mridul,
>
> I still have a question. DAGScheduler#submitMissingTasks will
> only unregisterAllMapAndMergeOutput
> if the current ShuffleMapStage is Indeterminate. What if the current stage
> is determinate, but its
> upstream stage is Indeterminate, and its upstream stage is rerun?
>
> Thanks,
> Keyong Zhou
>
> Mridul Muralidharan <mr...@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 <mr...@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 <zh...@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 <zh...@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 <fm...@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 <
>> mridul@gmail.com>
>> >> >> wrote:
>> >> >>
>> >> >> >
>> >> >> >
>> >> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <
>> >> mridul@gmail.com>
>> >> >> >> wrote:
>> >> >> >>
>> >> >> >>>
>> >> >> >>>
>> >> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <
>> >> mridul@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 <
>> glapark@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 <
>> >> >> mridul@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 <
>> >> >> orpl@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
>> >> >> >>>>>>>
>> >> >> >>>>>>
>> >> >>
>> >> >
>> >>
>> >
>>
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Keyong Zhou <zh...@apache.org>]

Hi Mridul,

I still have a question. DAGScheduler#submitMissingTasks will
only unregisterAllMapAndMergeOutput
if the current ShuffleMapStage is Indeterminate. What if the current stage
is determinate, but its
upstream stage is Indeterminate, and its upstream stage is rerun?

Thanks,
Keyong Zhou

Mridul Muralidharan <mr...@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 <mr...@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 <zh...@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 <zh...@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 <fm...@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 <
> mridul@gmail.com>
> >> >> wrote:
> >> >>
> >> >> >
> >> >> >
> >> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <
> >> mridul@gmail.com>
> >> >> >> wrote:
> >> >> >>
> >> >> >>>
> >> >> >>>
> >> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <
> >> mridul@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 <glapark@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 <
> >> >> mridul@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 <
> >> >> orpl@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
> >> >> >>>>>>>
> >> >> >>>>>>
> >> >>
> >> >
> >>
> >
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

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 <mr...@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 <zh...@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 <zh...@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 <fm...@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 <mr...@gmail.com>
>> >> wrote:
>> >>
>> >> >
>> >> >
>> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <
>> mridul@gmail.com>
>> >> >> wrote:
>> >> >>
>> >> >>>
>> >> >>>
>> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <
>> mridul@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 <gl...@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 <
>> >> mridul@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 <
>> >> orpl@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
>> >> >>>>>>>
>> >> >>>>>>
>> >>
>> >
>>
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

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 <zh...@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 <zh...@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 <fm...@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 <mr...@gmail.com>
> >> wrote:
> >>
> >> >
> >> >
> >> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <
> mridul@gmail.com>
> >> >> wrote:
> >> >>
> >> >>>
> >> >>>
> >> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <
> mridul@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 <gl...@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 <
> >> mridul@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 <
> >> orpl@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
> >> >>>>>>>
> >> >>>>>>
> >>
> >
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Keyong Zhou <zh...@apache.org>]

In fact, I'm wondering if Spark will rerun the whole reduce ShuffleMapStage
if its upstream ShuffleMapStage is INDETERMINATE and rerun.

Keyong Zhou <zh...@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 <fm...@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 <mr...@gmail.com>
>> wrote:
>>
>> >
>> >
>> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <mr...@gmail.com>
>> >> wrote:
>> >>
>> >>>
>> >>>
>> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <mridul@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 <gl...@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 <
>> mridul@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 <
>> orpl@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
>> >>>>>>>
>> >>>>>>
>>
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Keyong Zhou <zh...@apache.org>]

Thanks Erik for bringing up this question, I'm also curious about the
answer, any feedback is appreciated.

Thanks,
Keyong Zhou

Erik fang <fm...@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 <mr...@gmail.com>
> wrote:
>
> >
> >
> > On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <mr...@gmail.com>
> >> wrote:
> >>
> >>>
> >>>
> >>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <mr...@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 <gl...@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 <
> mridul@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 <orpl@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
> >>>>>>>
> >>>>>>
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Erik fang <fm...@gmail.com>]

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 <mr...@gmail.com>
wrote:

>
>
> On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <mr...@gmail.com>
>> wrote:
>>
>>>
>>>
>>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <mr...@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 <gl...@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 <mr...@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 <or...@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
>>>>>>>
>>>>>>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

On Mon, Oct 16, 2023 at 11:31 AM Erik fang <fm...@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 <mr...@gmail.com>
> wrote:
>
>>
>>
>> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <mr...@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 <gl...@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 <mr...@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 <or...@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
>>>>>>
>>>>>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Erik fang <fm...@gmail.com>]

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

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

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 <mr...@gmail.com>
wrote:

>
>
> On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <mr...@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 <gl...@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 <mr...@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 <or...@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
>>>>>
>>>>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

On Sat, Oct 14, 2023 at 3:49 AM Mridul Muralidharan <mr...@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 <gl...@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 <mr...@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 <or...@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
>>>>
>>>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

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 failure centric to Spark internals :-)

Regards,
Mridul


On Sat, Oct 14, 2023 at 3:36 AM Sungwoo Park <gl...@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 <mr...@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 <or...@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
>>>
>>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

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 <or...@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
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Sungwoo Park <or...@pl.postech.ac.kr>]

> 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

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Mridul Muralidharan <mr...@gmail.com>]

Hi,

  So there are a couple of things here based on whether the stages are
DETERMINATE or INDETERMINATE.
The exit I added to my example was to trigger some of these cases, and you
can come up with more involved scenarios where this would apply :-)

At a high level, we have the following:

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>
).
b) Conversely, if it was a DETERMINATE stage, only the missing partitions
for that shuffle id will be recomputed.
c) This also impacts whether we can use a task's output if it was for an
earlier stage attempt or not - we can use it for an older attempt if
DETERMINATE, and cant for INDETERMINATE (see here
<https://github.com/apache/spark/blob/3e2470de7ea8b97dcdd8875ef25f044998fb7588/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1852>,
here
<https://github.com/apache/spark/blob/3e2470de7ea8b97dcdd8875ef25f044998fb7588/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala#L1907>
).
(We can skip the more involved cases, should not be relevant for Celeborn
IMO).

For a given partition P, for a spark-shuffle-id, MapOutputTracker state is
updated by DAGScheduler keeping these in mind.
Assuming I did not misunderstand the proposal, if each stage attempt
corresponds to a celeborn-shuffle-id, such that we have spark-shuffle-id ->
List[celeborn-shuffle-id], we will need to take these cases above into
account as well when a "reducer" is requesting for spark-shuffle-id,
partition P output (which celeborn-shuffle-id it maps to).


Regards,
Mridul





On Thu, Oct 12, 2023 at 3:14 AM Erik fang <fm...@gmail.com> wrote:

> Hi Mridul,
>
> sorry for the late reply
>
> Per my understanding, the key point about Spark shuffleId and
> StageId/StageAttemptId is,
> shuffleId is assigned at ShuffleDependency creation time and bounded to
> the RDD/ShuffleDependency, while StageId/StageAttemptId is assigned and
> changes at Job execution time
> In the RDD example, there are two shuffle data with id 0 and 1,  and those
> shuffle data are expected to be accessed(read/write) with the correct
> shuffle id
> and we can do that in celeborn with shuffle id mapping
>
> I made some small modification to the example to avoid exit , and grab
> some logs for Spark DAGScheduler/Celeborn LifecycleManager to help explain
>
> import org.apache.spark.TaskContext
>
> val rdd1 = sc.parallelize(0 until 10000, 20).map(v => (v, v)).groupByKey()
> val rdd2 = rdd1.mapPartitions { iter =>
>   val tc = TaskContext.get()
>   println("print stageAttemptNumber " + tc.stageAttemptNumber())
>   iter
> }
>
> rdd2.count()
> rdd2.map(v => (v._1, v._2)).groupByKey().count()
>
> *// DAGScheduler starts job 0, submit ShuffleMapStage 0 for spark
> shuffle-0*
> 23/10/12 12:58:14 INFO DAGScheduler: Registering RDD 1 (map at <console>:24)
> as input to shuffle 0
> 23/10/12 12:58:14 INFO DAGScheduler: Got job 0 (count at <console>:25)
> with 20 output partitions
> 23/10/12 12:58:14 INFO DAGScheduler: Final stage: ResultStage 1 (count at
> <console>:25)
> 23/10/12 12:58:14 INFO DAGScheduler: Parents of final stage:
> List(ShuffleMapStage 0)
> 23/10/12 12:58:14 INFO DAGScheduler: Missing parents: List(ShuffleMapStage
> 0)
> 23/10/12 12:58:14 DEBUG DAGScheduler: submitStage(ResultStage 1
> (name=count at <console>:25;jobs=0))
> 23/10/12 12:58:14 DEBUG DAGScheduler: missing: List(ShuffleMapStage 0)
> 23/10/12 12:58:14 DEBUG DAGScheduler: submitStage(ShuffleMapStage 0
> (name=map at <console>:24;jobs=0))
> 23/10/12 12:58:14 DEBUG DAGScheduler: missing: List()
> 23/10/12 12:58:14 INFO DAGScheduler: Submitting ShuffleMapStage 0
> (MapPartitionsRDD[1] at map at <console>:24), which has no missing parents
> 23/10/12 12:58:14 DEBUG DAGScheduler: submitMissingTasks(ShuffleMapStage 0)
>
> *// LifecycleManager received GetShuffleId request from ShuffleMapStage 0
> with spark_shuffleId 0, stage attemptId 0, and generate celeborn_shuffleId
> 0 for write*
> 23/10/12 12:58:16 DEBUG LifecycleManager: Received GetShuffleId
> request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter true
> 23/10/12 12:58:16 DEBUG LifecycleManager: Received RegisterShuffle
> request, 0, 20, 20.
> 23/10/12 12:58:16 INFO LifecycleManager: New shuffle request, shuffleId 0,
> partitionType: REDUCE numMappers: 20, numReducers: 20.
>
> *// ShuffleMapStage finish, Submit ResultStage 1*
> 23/10/12 12:58:18 INFO LifecycleManager: Received StageEnd request,
> shuffleId 0.
> 23/10/12 12:58:18 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks
> have all completed, from pool
> 23/10/12 12:58:18 INFO DAGScheduler: ShuffleMapStage 0 (map at <console>:24)
> finished in 4.246 s
> 23/10/12 12:58:18 INFO DAGScheduler: looking for newly runnable stages
> 23/10/12 12:58:18 INFO DAGScheduler: running: Set()
> 23/10/12 12:58:18 INFO DAGScheduler: waiting: Set(ResultStage 1)
> 23/10/12 12:58:18 INFO DAGScheduler: failed: Set()
> 23/10/12 12:58:18 DEBUG DAGScheduler: submitStage(ResultStage 1
> (name=count at <console>:25;jobs=0))
> 23/10/12 12:58:18 DEBUG DAGScheduler: missing: List()
> 23/10/12 12:58:18 INFO DAGScheduler: Submitting ResultStage 1
> (MapPartitionsRDD[3] at mapPartitions at <console>:24), which has no
> missing parents
> 23/10/12 12:58:18 DEBUG DAGScheduler: submitMissingTasks(ResultStage 1)
>
> *// LifecycleManager received GetShuffleId request from ResultStage 1 with
> spark_shuffleId 0, stage attemptId 0*
> *// and generate celeborn_shuffleId 0 to return shuffle metadata with
> celeborn's GetShuffleFileGroup for read*
> 23/10/12 12:58:19 DEBUG LifecycleManager: Received GetShuffleId
> request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter false
> 23/10/12 12:58:19 DEBUG CelebornShuffleReader: read shuffleId 0 for
> appShuffleId 0 attemptNum 0
> 23/10/12 12:58:19 DEBUG LifecycleManager: Received GetShuffleFileGroup
> request,shuffleId 0.
>
> 23/10/12 12:58:20 INFO DAGScheduler: ResultStage 1 (count at <console>:25)
> finished in 1.546 s
> 23/10/12 12:58:20 INFO DAGScheduler: Job 0 is finished. Cancelling
> potential speculative or zombie tasks for this job
>
> *// DAGScheduler starts job 1, submit ShuffleMapStage 3 for spark
> shuffle-1*
> 23/10/12 12:58:22 INFO DAGScheduler: Registering RDD 4 (map at <console>:25)
> as input to shuffle 1
> 23/10/12 12:58:22 INFO DAGScheduler: Got job 1 (count at <console>:25)
> with 20 output partitions
> 23/10/12 12:58:22 INFO DAGScheduler: Final stage: ResultStage 4 (count at
> <console>:25)
> 23/10/12 12:58:22 INFO DAGScheduler: Parents of final stage:
> List(ShuffleMapStage 3)
> 23/10/12 12:58:22 INFO DAGScheduler: Missing parents: List(ShuffleMapStage
> 3)
> 23/10/12 12:58:22 DEBUG DAGScheduler: submitStage(ResultStage 4
> (name=count at <console>:25;jobs=1))
> 23/10/12 12:58:22 DEBUG DAGScheduler: missing: List(ShuffleMapStage 3)
> 23/10/12 12:58:22 DEBUG DAGScheduler: submitStage(ShuffleMapStage 3
> (name=map at <console>:25;jobs=1))
> 23/10/12 12:58:22 DEBUG DAGScheduler: missing: List()
> 23/10/12 12:58:22 INFO DAGScheduler: Submitting ShuffleMapStage 3
> (MapPartitionsRDD[4] at map at <console>:25), which has no missing parents
> 23/10/12 12:58:22 DEBUG DAGScheduler: submitMissingTasks(ShuffleMapStage 3)
>
> *// LifecycleManager Received GetShuffleId request from ShuffleMapStage 3
> with spark_shuffleId 1, stage attemptId 0， generate celeborn_shuffleId 4
> for write*
> 23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
> request,appShuffleId 1 maxAttemptNum 4 attemptId 0 isShuffleWriter true
> 23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
> request,appShuffleId 1 maxAttemptNum 4 attemptId 0 isShuffleWriter true
>
> *// LifecycleManager Received GetShuffleId request from ShuffleMapStage 3
> with spark_shuffleId 0, stage attemptId 0 to read spark shuffle-0*
> 23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
> request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter false
> 23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
> request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter false
> 23/10/12 12:58:22 DEBUG CelebornShuffleReader: read shuffleId 0 for
> appShuffleId 0 attemptNum 0
>
> 23/10/12 12:58:22 DEBUG LifecycleManager: Received RegisterShuffle
> request, 4, 20, 20.
> 23/10/12 12:58:22 INFO LifecycleManager: New shuffle request, shuffleId 4,
> partitionType: REDUCE numMappers: 20, numReducers: 20.
>
> 23/10/12 12:58:24 INFO DAGScheduler: ShuffleMapStage 3 (map at <console>:25)
> finished in 2.359 s
> 23/10/12 12:58:24 INFO DAGScheduler: looking for newly runnable stages
> 23/10/12 12:58:24 INFO DAGScheduler: running: Set()
> 23/10/12 12:58:24 INFO DAGScheduler: waiting: Set(ResultStage 4)
> 23/10/12 12:58:24 INFO DAGScheduler: failed: Set()
> 23/10/12 12:58:24 DEBUG DAGScheduler: submitStage(ResultStage 4
> (name=count at <console>:25;jobs=1))
> 23/10/12 12:58:24 DEBUG DAGScheduler: missing: List()
> 23/10/12 12:58:24 INFO DAGScheduler: Submitting ResultStage 4 (ShuffledRDD[
> 5] at groupByKey at <console>:25), which has no missing parents
>
> we can see that both jobs access the first shuffle data with correct
> shuffle id
>
> However, you comments remind me one missing case here:
> If shuffle 0 failed with first stage attempt, then succeeded with second
> attempt, job 1 can get correct result (read with spark shuffle id 0, stage
> attempt id 1, thus celeborn shuffle id 1)
> For job 2, it gets incorrect celeborn shuffle id when read shuffle 0 for
> the first stage attempts (read with spark shuffle id 0, stage attempt id 0,
> thus celeborn shuffle id 0, but we need celeborn shuffle id 1 here)
> To fix this, LifecycleManager needs to keep the mapping of Spark shuffle
> id -> final success celeborn shuffle id across Stage attempts, and return
> celeborn shuffle id to shuffle reader
>
> Thanks,
> Erik
>
> On Sun, Oct 8, 2023 at 3:12 PM Sungwoo Park <or...@pl.postech.ac.kr> wrote:
>
>> Hello Keyong,
>>
>> We have implemented the previous plan to generate a fresh shufflId when
>> failing to fetch data from Celeborn workers. Thanks for your comment.
>>
>> While testing Celeborn-MR3 with task re-execution (or stage
>> resubmission),
>> I have noticed that shuffleId does not change its associated set of
>> Celeborn workers. Specifically, pushData() for a specific shuffleId always
>> fails if any of its associated Celeborn workers is killed. Our experiment
>> goes as follows:
>>
>> 1. Execute a long-running query.
>> 2. While the query is running (many mappers have succeeded and some
>> reducers are still running), kill 2 Celeborn workers out of 12.
>> 3. A task fails after three attempts with this stack trace:
>>
>> 2023-10-08 05:23:43,814 [DAG-2-23-2] ERROR
>> org.apache.celeborn.client.ShuffleClientImpl [] - Exception raised while
>> pushing data for shuffle 25 map 113 attempt 3 partition 0 batch 1
>> location
>> PartitionLocation[
>>    id-epoch:0-0
>>
>>
>> host-rpcPort-pushPort-fetchPort-replicatePort:192.168.10.113-45109-37140-45752-40163
>>    mode:PRIMARY
>>    peer:(empty)
>>    storage hint:StorageInfo{type=MEMORY, mountPoint='/data3',
>> finalResult=false, filePath=}
>>    mapIdBitMap:null].
>> org.apache.celeborn.common.exception.CelebornIOException: Failed to
>> connect to /192.168.10.113:37140
>>    at
>> org.apache.celeborn.common.network.client.TransportClientFactory.internalCreateClient(TransportClientFactory.java:246)
>>
>> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>>    at
>> org.apache.celeborn.common.network.client.TransportClientFactory.createClient(TransportClientFactory.java:190)
>>
>> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>>    at
>> org.apache.celeborn.common.network.client.TransportClientFactory.createClient(TransportClientFactory.java:116)
>>
>> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>>    at
>> org.apache.celeborn.client.ShuffleClientImpl.pushOrMergeData(ShuffleClientImpl.java:1079)
>>
>> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>>    at
>> org.apache.celeborn.client.ShuffleClientImpl.pushData(ShuffleClientImpl.java:1162)
>>
>> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>>    at
>> org.apache.tez.runtime.library.common.sort.impl.PipelinedSorter.doFinalMerge(PipelinedSorter.java:979)
>>
>> ~[tez-runtime-library-0.9.1.mr3.1.0.jar:0.9.1.mr3.1.0]
>>
>> So, does this mean that if pushData() fails with this exception (after
>> some Celeborn workers are lost):
>>
>>      org.apache.celeborn.common.exception.CelebornIOException: Failed to
>> connect to /192.168.10.113:37140
>>
>> we should stop using the current shuffleId and use a fresh shuffleId
>> instead? (I guess the same phenomenon will occur for stage resubmission
>> in
>> Spark.)
>>
>> Or, generating a fresh shuffleId does not help because it may still use
>> those lost Celeborn workers?
>>
>> Thanks,
>>
>> --- Sungwoo
>>
>> On Sat, 7 Oct 2023, Keyong Zhou wrote:
>>
>> > Hi Sungwoo,
>> >
>> > Sorry for the late reply. Reusing a committed shuffleId does not work in
>> > current architecture, even after calling unregisterShuffle in
>> > LifecycleManager,
>> > because the cleanup of metadata is delayed and not guaranteed. It will
>> > be more complicated when we consider graceful restart of workers.
>> >
>> > If we want to reuse the shuffleId, we need to redesign the whole
>> picture.
>> >
>> > Thanks,
>> > Keyong Zhou
>> >
>> > Sungwoo Park <or...@pl.postech.ac.kr> 于2023年10月2日周一 13:23?道：
>> >
>> >> Hi Keyong,
>> >>
>> >> Instead of picking up a new shuffleId, can we reuse an existing
>> shuffleId
>> >> after unregistering it? If the following plan worked, it would further
>> >> simplify the implementation:
>> >>
>> >> 1. Downstream tasks fail because of read failures.
>> >> 2. All active downstream tasks are killed, so the shuffleId is not
>> used.
>> >> 3. An upstream vertex unregisters the shuffleId.
>> >> 4. The upstream vertex is re-executed normally. This re-execution
>> >> automatically registers the same shuffleId.
>> >>
>> >> In summary, we would like to go back in time before the upstream vertex
>> >> started by cleaning up the shuffleId. Could you please give a comment
>> on
>> >> this plan?
>> >>
>> >> Thank you,
>> >>
>> >> --- Sungwoo
>> >>
>> >> On Sat, 30 Sep 2023, Keyong Zhou wrote:
>> >>
>> >>> Hi Sungwoo,
>> >>>
>> >>> I think your approach works with current architecture of Celeborn,
>> >>> and interpreting IOException when reading as read failure makes
>> >>> sense. Currently only when CommitFiles fails will LifecycleManager
>> >>> announce data lost.
>> >>>
>> >>> Thanks,
>> >>> Keyong Zhou
>> >>>
>> >>> Sungwoo Park <or...@pl.postech.ac.kr> 于2023年9月29日周五 22:05?道：
>> >>>
>> >>>>> Since the partition split has a good chance to contain data from
>> almost
>> >>>> all
>> >>>>> upstream
>> >>>>> mapper tasks, the cost of re-computing all upstream tasks may have
>> >> little
>> >>>>> difference
>> >>>>> to re-computing the actual mapper tasks in most cases. Of course
>> it's
>> >> not
>> >>>>> always true.
>> >>>>>
>> >>>>> To change from 'complete' to 'incomplete' also needs to refactor
>> >> Worker's
>> >>>>> logic, which
>> >>>>> currently assumes that the succeeded attempts will not be changed
>> after
>> >>>>> final committing
>> >>>>> files.
>> >>>>>
>> >>>>>> a subset of succeeded attempts. In Erik's proposal, the whole
>> upstream
>> >>>>>> stage will be rerun when data lost.
>> >>>>
>> >>>> Thank you for your response --- things are now much clearer.
>> >>>>
>> >>>> From your comments shown above, let me assume that:
>> >>>>
>> >>>> 1. The whole upstream stage is rerun in the case of read failure.
>> >>>>
>> >>>> 2. Currently it is not easy to change the state of a shuffleId from
>> >>>> 'complete' to 'incomplete'.
>> >>>>
>> >>>> Then, for Celeborn-MR3, I would like to experiment with the following
>> >>>> approach:
>> >>>>
>> >>>> 1. If read failures occur for shuffleId #1, we pick up a new
>> shuffleId
>> >>>> #2.
>> >>>>
>> >>>> 2. The upstream stage (or Vertex in the case of MR3) re-executes all
>> >> tasks
>> >>>> again, but writes the output to shuffleId #2.
>> >>>>
>> >>>> 3. Tasks in the downstream stage re-try by reading from shuffleId #2.
>> >>>>
>> >>>> Do you think this approach makes sense under the current
>> architecture of
>> >>>> Celeborn? If this approach is feasible, MR3 only needs to be
>> notified of
>> >>>> read failures due to lost data by Celeborn ShuffleClient. Or, we
>> could
>> >>>> just interpret IOException from Celeborn ShuffleClient as read
>> failures,
>> >>>> in which case we can implement stage recompute without requiring any
>> >>>> extension of Celeborn. (However, it would be great if Celeborn
>> >>>> ShuffleClient could announce lost data explicitly.)
>> >>>>
>> >>>> An industrial user of Hive-MR3-Celeborn is trying hard to save disk
>> >> usage
>> >>>> on Celeborn workers (which use SSDs of limited capacity), so stage
>> >>>> recompute would be a great new feature to them.
>> >>>>
>> >>>> Thank you,
>> >>>>
>> >>>> --- Sungwoo
>> >>>>
>> >>>>
>> >>>>
>> >>>
>> >
>
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Erik fang <fm...@gmail.com>]

Hi Mridul,

sorry for the late reply

Per my understanding, the key point about Spark shuffleId and
StageId/StageAttemptId is,
shuffleId is assigned at ShuffleDependency creation time and bounded to the
RDD/ShuffleDependency, while StageId/StageAttemptId is assigned and changes
at Job execution time
In the RDD example, there are two shuffle data with id 0 and 1,  and those
shuffle data are expected to be accessed(read/write) with the correct
shuffle id
and we can do that in celeborn with shuffle id mapping

I made some small modification to the example to avoid exit , and grab some
logs for Spark DAGScheduler/Celeborn LifecycleManager to help explain

import org.apache.spark.TaskContext

val rdd1 = sc.parallelize(0 until 10000, 20).map(v => (v, v)).groupByKey()
val rdd2 = rdd1.mapPartitions { iter =>
  val tc = TaskContext.get()
  println("print stageAttemptNumber " + tc.stageAttemptNumber())
  iter
}

rdd2.count()
rdd2.map(v => (v._1, v._2)).groupByKey().count()

*// DAGScheduler starts job 0, submit ShuffleMapStage 0 for spark shuffle-0*
23/10/12 12:58:14 INFO DAGScheduler: Registering RDD 1 (map at <console>:24)
as input to shuffle 0
23/10/12 12:58:14 INFO DAGScheduler: Got job 0 (count at <console>:25) with
20 output partitions
23/10/12 12:58:14 INFO DAGScheduler: Final stage: ResultStage 1 (count at <
console>:25)
23/10/12 12:58:14 INFO DAGScheduler: Parents of final stage:
List(ShuffleMapStage 0)
23/10/12 12:58:14 INFO DAGScheduler: Missing parents: List(ShuffleMapStage
0)
23/10/12 12:58:14 DEBUG DAGScheduler: submitStage(ResultStage 1 (name=count
at <console>:25;jobs=0))
23/10/12 12:58:14 DEBUG DAGScheduler: missing: List(ShuffleMapStage 0)
23/10/12 12:58:14 DEBUG DAGScheduler: submitStage(ShuffleMapStage 0
(name=map at <console>:24;jobs=0))
23/10/12 12:58:14 DEBUG DAGScheduler: missing: List()
23/10/12 12:58:14 INFO DAGScheduler: Submitting ShuffleMapStage 0
(MapPartitionsRDD[1] at map at <console>:24), which has no missing parents
23/10/12 12:58:14 DEBUG DAGScheduler: submitMissingTasks(ShuffleMapStage 0)

*// LifecycleManager received GetShuffleId request from ShuffleMapStage 0
with spark_shuffleId 0, stage attemptId 0, and generate celeborn_shuffleId
0 for write*
23/10/12 12:58:16 DEBUG LifecycleManager: Received GetShuffleId
request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter true
23/10/12 12:58:16 DEBUG LifecycleManager: Received RegisterShuffle request,
0, 20, 20.
23/10/12 12:58:16 INFO LifecycleManager: New shuffle request, shuffleId 0,
partitionType: REDUCE numMappers: 20, numReducers: 20.

*// ShuffleMapStage finish, Submit ResultStage 1*
23/10/12 12:58:18 INFO LifecycleManager: Received StageEnd request,
shuffleId 0.
23/10/12 12:58:18 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks
have all completed, from pool
23/10/12 12:58:18 INFO DAGScheduler: ShuffleMapStage 0 (map at <console>:24)
finished in 4.246 s
23/10/12 12:58:18 INFO DAGScheduler: looking for newly runnable stages
23/10/12 12:58:18 INFO DAGScheduler: running: Set()
23/10/12 12:58:18 INFO DAGScheduler: waiting: Set(ResultStage 1)
23/10/12 12:58:18 INFO DAGScheduler: failed: Set()
23/10/12 12:58:18 DEBUG DAGScheduler: submitStage(ResultStage 1 (name=count
at <console>:25;jobs=0))
23/10/12 12:58:18 DEBUG DAGScheduler: missing: List()
23/10/12 12:58:18 INFO DAGScheduler: Submitting ResultStage 1
(MapPartitionsRDD[3] at mapPartitions at <console>:24), which has no
missing parents
23/10/12 12:58:18 DEBUG DAGScheduler: submitMissingTasks(ResultStage 1)

*// LifecycleManager received GetShuffleId request from ResultStage 1 with
spark_shuffleId 0, stage attemptId 0*
*// and generate celeborn_shuffleId 0 to return shuffle metadata with
celeborn's GetShuffleFileGroup for read*
23/10/12 12:58:19 DEBUG LifecycleManager: Received GetShuffleId
request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter false
23/10/12 12:58:19 DEBUG CelebornShuffleReader: read shuffleId 0 for
appShuffleId 0 attemptNum 0
23/10/12 12:58:19 DEBUG LifecycleManager: Received GetShuffleFileGroup
request,shuffleId 0.

23/10/12 12:58:20 INFO DAGScheduler: ResultStage 1 (count at <console>:25)
finished in 1.546 s
23/10/12 12:58:20 INFO DAGScheduler: Job 0 is finished. Cancelling
potential speculative or zombie tasks for this job

*// DAGScheduler starts job 1, submit ShuffleMapStage 3 for spark shuffle-1*
23/10/12 12:58:22 INFO DAGScheduler: Registering RDD 4 (map at <console>:25)
as input to shuffle 1
23/10/12 12:58:22 INFO DAGScheduler: Got job 1 (count at <console>:25) with
20 output partitions
23/10/12 12:58:22 INFO DAGScheduler: Final stage: ResultStage 4 (count at <
console>:25)
23/10/12 12:58:22 INFO DAGScheduler: Parents of final stage:
List(ShuffleMapStage 3)
23/10/12 12:58:22 INFO DAGScheduler: Missing parents: List(ShuffleMapStage
3)
23/10/12 12:58:22 DEBUG DAGScheduler: submitStage(ResultStage 4 (name=count
at <console>:25;jobs=1))
23/10/12 12:58:22 DEBUG DAGScheduler: missing: List(ShuffleMapStage 3)
23/10/12 12:58:22 DEBUG DAGScheduler: submitStage(ShuffleMapStage 3
(name=map at <console>:25;jobs=1))
23/10/12 12:58:22 DEBUG DAGScheduler: missing: List()
23/10/12 12:58:22 INFO DAGScheduler: Submitting ShuffleMapStage 3
(MapPartitionsRDD[4] at map at <console>:25), which has no missing parents
23/10/12 12:58:22 DEBUG DAGScheduler: submitMissingTasks(ShuffleMapStage 3)

*// LifecycleManager Received GetShuffleId request from ShuffleMapStage 3
with spark_shuffleId 1, stage attemptId 0， generate celeborn_shuffleId 4
for write*
23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
request,appShuffleId 1 maxAttemptNum 4 attemptId 0 isShuffleWriter true
23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
request,appShuffleId 1 maxAttemptNum 4 attemptId 0 isShuffleWriter true

*// LifecycleManager Received GetShuffleId request from ShuffleMapStage 3
with spark_shuffleId 0, stage attemptId 0 to read spark shuffle-0*
23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter false
23/10/12 12:58:22 DEBUG LifecycleManager: Received GetShuffleId
request,appShuffleId 0 maxAttemptNum 4 attemptId 0 isShuffleWriter false
23/10/12 12:58:22 DEBUG CelebornShuffleReader: read shuffleId 0 for
appShuffleId 0 attemptNum 0

23/10/12 12:58:22 DEBUG LifecycleManager: Received RegisterShuffle request,
4, 20, 20.
23/10/12 12:58:22 INFO LifecycleManager: New shuffle request, shuffleId 4,
partitionType: REDUCE numMappers: 20, numReducers: 20.

23/10/12 12:58:24 INFO DAGScheduler: ShuffleMapStage 3 (map at <console>:25)
finished in 2.359 s
23/10/12 12:58:24 INFO DAGScheduler: looking for newly runnable stages
23/10/12 12:58:24 INFO DAGScheduler: running: Set()
23/10/12 12:58:24 INFO DAGScheduler: waiting: Set(ResultStage 4)
23/10/12 12:58:24 INFO DAGScheduler: failed: Set()
23/10/12 12:58:24 DEBUG DAGScheduler: submitStage(ResultStage 4 (name=count
at <console>:25;jobs=1))
23/10/12 12:58:24 DEBUG DAGScheduler: missing: List()
23/10/12 12:58:24 INFO DAGScheduler: Submitting ResultStage 4 (ShuffledRDD[5]
at groupByKey at <console>:25), which has no missing parents

we can see that both jobs access the first shuffle data with correct
shuffle id

However, you comments remind me one missing case here:
If shuffle 0 failed with first stage attempt, then succeeded with second
attempt, job 1 can get correct result (read with spark shuffle id 0, stage
attempt id 1, thus celeborn shuffle id 1)
For job 2, it gets incorrect celeborn shuffle id when read shuffle 0 for
the first stage attempts (read with spark shuffle id 0, stage attempt id 0,
thus celeborn shuffle id 0, but we need celeborn shuffle id 1 here)
To fix this, LifecycleManager needs to keep the mapping of Spark shuffle id
-> final success celeborn shuffle id across Stage attempts, and return
celeborn shuffle id to shuffle reader

Thanks,
Erik

On Sun, Oct 8, 2023 at 3:12 PM Sungwoo Park <or...@pl.postech.ac.kr> wrote:

> Hello Keyong,
>
> We have implemented the previous plan to generate a fresh shufflId when
> failing to fetch data from Celeborn workers. Thanks for your comment.
>
> While testing Celeborn-MR3 with task re-execution (or stage resubmission),
> I have noticed that shuffleId does not change its associated set of
> Celeborn workers. Specifically, pushData() for a specific shuffleId always
> fails if any of its associated Celeborn workers is killed. Our experiment
> goes as follows:
>
> 1. Execute a long-running query.
> 2. While the query is running (many mappers have succeeded and some
> reducers are still running), kill 2 Celeborn workers out of 12.
> 3. A task fails after three attempts with this stack trace:
>
> 2023-10-08 05:23:43,814 [DAG-2-23-2] ERROR
> org.apache.celeborn.client.ShuffleClientImpl [] - Exception raised while
> pushing data for shuffle 25 map 113 attempt 3 partition 0 batch 1 location
> PartitionLocation[
>    id-epoch:0-0
>
>
> host-rpcPort-pushPort-fetchPort-replicatePort:192.168.10.113-45109-37140-45752-40163
>    mode:PRIMARY
>    peer:(empty)
>    storage hint:StorageInfo{type=MEMORY, mountPoint='/data3',
> finalResult=false, filePath=}
>    mapIdBitMap:null].
> org.apache.celeborn.common.exception.CelebornIOException: Failed to
> connect to /192.168.10.113:37140
>    at
> org.apache.celeborn.common.network.client.TransportClientFactory.internalCreateClient(TransportClientFactory.java:246)
>
> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>    at
> org.apache.celeborn.common.network.client.TransportClientFactory.createClient(TransportClientFactory.java:190)
>
> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>    at
> org.apache.celeborn.common.network.client.TransportClientFactory.createClient(TransportClientFactory.java:116)
>
> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>    at
> org.apache.celeborn.client.ShuffleClientImpl.pushOrMergeData(ShuffleClientImpl.java:1079)
>
> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>    at
> org.apache.celeborn.client.ShuffleClientImpl.pushData(ShuffleClientImpl.java:1162)
>
> ~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
>    at
> org.apache.tez.runtime.library.common.sort.impl.PipelinedSorter.doFinalMerge(PipelinedSorter.java:979)
>
> ~[tez-runtime-library-0.9.1.mr3.1.0.jar:0.9.1.mr3.1.0]
>
> So, does this mean that if pushData() fails with this exception (after
> some Celeborn workers are lost):
>
>      org.apache.celeborn.common.exception.CelebornIOException: Failed to
> connect to /192.168.10.113:37140
>
> we should stop using the current shuffleId and use a fresh shuffleId
> instead? (I guess the same phenomenon will occur for stage resubmission in
> Spark.)
>
> Or, generating a fresh shuffleId does not help because it may still use
> those lost Celeborn workers?
>
> Thanks,
>
> --- Sungwoo
>
> On Sat, 7 Oct 2023, Keyong Zhou wrote:
>
> > Hi Sungwoo,
> >
> > Sorry for the late reply. Reusing a committed shuffleId does not work in
> > current architecture, even after calling unregisterShuffle in
> > LifecycleManager,
> > because the cleanup of metadata is delayed and not guaranteed. It will
> > be more complicated when we consider graceful restart of workers.
> >
> > If we want to reuse the shuffleId, we need to redesign the whole picture.
> >
> > Thanks,
> > Keyong Zhou
> >
> > Sungwoo Park <or...@pl.postech.ac.kr> 于2023年10月2日周一 13:23?道：
> >
> >> Hi Keyong,
> >>
> >> Instead of picking up a new shuffleId, can we reuse an existing
> shuffleId
> >> after unregistering it? If the following plan worked, it would further
> >> simplify the implementation:
> >>
> >> 1. Downstream tasks fail because of read failures.
> >> 2. All active downstream tasks are killed, so the shuffleId is not used.
> >> 3. An upstream vertex unregisters the shuffleId.
> >> 4. The upstream vertex is re-executed normally. This re-execution
> >> automatically registers the same shuffleId.
> >>
> >> In summary, we would like to go back in time before the upstream vertex
> >> started by cleaning up the shuffleId. Could you please give a comment on
> >> this plan?
> >>
> >> Thank you,
> >>
> >> --- Sungwoo
> >>
> >> On Sat, 30 Sep 2023, Keyong Zhou wrote:
> >>
> >>> Hi Sungwoo,
> >>>
> >>> I think your approach works with current architecture of Celeborn,
> >>> and interpreting IOException when reading as read failure makes
> >>> sense. Currently only when CommitFiles fails will LifecycleManager
> >>> announce data lost.
> >>>
> >>> Thanks,
> >>> Keyong Zhou
> >>>
> >>> Sungwoo Park <or...@pl.postech.ac.kr> 于2023年9月29日周五 22:05?道：
> >>>
> >>>>> Since the partition split has a good chance to contain data from
> almost
> >>>> all
> >>>>> upstream
> >>>>> mapper tasks, the cost of re-computing all upstream tasks may have
> >> little
> >>>>> difference
> >>>>> to re-computing the actual mapper tasks in most cases. Of course it's
> >> not
> >>>>> always true.
> >>>>>
> >>>>> To change from 'complete' to 'incomplete' also needs to refactor
> >> Worker's
> >>>>> logic, which
> >>>>> currently assumes that the succeeded attempts will not be changed
> after
> >>>>> final committing
> >>>>> files.
> >>>>>
> >>>>>> a subset of succeeded attempts. In Erik's proposal, the whole
> upstream
> >>>>>> stage will be rerun when data lost.
> >>>>
> >>>> Thank you for your response --- things are now much clearer.
> >>>>
> >>>> From your comments shown above, let me assume that:
> >>>>
> >>>> 1. The whole upstream stage is rerun in the case of read failure.
> >>>>
> >>>> 2. Currently it is not easy to change the state of a shuffleId from
> >>>> 'complete' to 'incomplete'.
> >>>>
> >>>> Then, for Celeborn-MR3, I would like to experiment with the following
> >>>> approach:
> >>>>
> >>>> 1. If read failures occur for shuffleId #1, we pick up a new shuffleId
> >>>> #2.
> >>>>
> >>>> 2. The upstream stage (or Vertex in the case of MR3) re-executes all
> >> tasks
> >>>> again, but writes the output to shuffleId #2.
> >>>>
> >>>> 3. Tasks in the downstream stage re-try by reading from shuffleId #2.
> >>>>
> >>>> Do you think this approach makes sense under the current architecture
> of
> >>>> Celeborn? If this approach is feasible, MR3 only needs to be notified
> of
> >>>> read failures due to lost data by Celeborn ShuffleClient. Or, we could
> >>>> just interpret IOException from Celeborn ShuffleClient as read
> failures,
> >>>> in which case we can implement stage recompute without requiring any
> >>>> extension of Celeborn. (However, it would be great if Celeborn
> >>>> ShuffleClient could announce lost data explicitly.)
> >>>>
> >>>> An industrial user of Hive-MR3-Celeborn is trying hard to save disk
> >> usage
> >>>> on Celeborn workers (which use SSDs of limited capacity), so stage
> >>>> recompute would be a great new feature to them.
> >>>>
> >>>> Thank you,
> >>>>
> >>>> --- Sungwoo
> >>>>
> >>>>
> >>>>
> >>>
> >

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Sungwoo Park <or...@pl.postech.ac.kr>]

Hello Keyong,

We have implemented the previous plan to generate a fresh shufflId when 
failing to fetch data from Celeborn workers. Thanks for your comment.

While testing Celeborn-MR3 with task re-execution (or stage resubmission), 
I have noticed that shuffleId does not change its associated set of 
Celeborn workers. Specifically, pushData() for a specific shuffleId always
fails if any of its associated Celeborn workers is killed. Our experiment
goes as follows:

1. Execute a long-running query.
2. While the query is running (many mappers have succeeded and some 
reducers are still running), kill 2 Celeborn workers out of 12.
3. A task fails after three attempts with this stack trace:

2023-10-08 05:23:43,814 [DAG-2-23-2] ERROR 
org.apache.celeborn.client.ShuffleClientImpl [] - Exception raised while 
pushing data for shuffle 25 map 113 attempt 3 partition 0 batch 1 location 
PartitionLocation[
   id-epoch:0-0

host-rpcPort-pushPort-fetchPort-replicatePort:192.168.10.113-45109-37140-45752-40163
   mode:PRIMARY
   peer:(empty)
   storage hint:StorageInfo{type=MEMORY, mountPoint='/data3', 
finalResult=false, filePath=}
   mapIdBitMap:null].
org.apache.celeborn.common.exception.CelebornIOException: Failed to 
connect to /192.168.10.113:37140
   at 
org.apache.celeborn.common.network.client.TransportClientFactory.internalCreateClient(TransportClientFactory.java:246) 
~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
   at 
org.apache.celeborn.common.network.client.TransportClientFactory.createClient(TransportClientFactory.java:190) 
~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
   at 
org.apache.celeborn.common.network.client.TransportClientFactory.createClient(TransportClientFactory.java:116) 
~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
   at 
org.apache.celeborn.client.ShuffleClientImpl.pushOrMergeData(ShuffleClientImpl.java:1079) 
~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
   at 
org.apache.celeborn.client.ShuffleClientImpl.pushData(ShuffleClientImpl.java:1162) 
~[celeborn-client-mr3-1.8-shaded_2.12-0.4.0-SNAPSHOT.jar:?]
   at 
org.apache.tez.runtime.library.common.sort.impl.PipelinedSorter.doFinalMerge(PipelinedSorter.java:979) 
~[tez-runtime-library-0.9.1.mr3.1.0.jar:0.9.1.mr3.1.0]

So, does this mean that if pushData() fails with this exception (after 
some Celeborn workers are lost):

     org.apache.celeborn.common.exception.CelebornIOException: Failed to connect to /192.168.10.113:37140

we should stop using the current shuffleId and use a fresh shuffleId 
instead? (I guess the same phenomenon will occur for stage resubmission in 
Spark.)

Or, generating a fresh shuffleId does not help because it may still use 
those lost Celeborn workers?

Thanks,

--- Sungwoo

On Sat, 7 Oct 2023, Keyong Zhou wrote:

> Hi Sungwoo,
>
> Sorry for the late reply. Reusing a committed shuffleId does not work in
> current architecture, even after calling unregisterShuffle in
> LifecycleManager,
> because the cleanup of metadata is delayed and not guaranteed. It will
> be more complicated when we consider graceful restart of workers.
>
> If we want to reuse the shuffleId, we need to redesign the whole picture.
>
> Thanks,
> Keyong Zhou
>
> Sungwoo Park <or...@pl.postech.ac.kr> 于2023年10月2日周一 13:23?道：
>
>> Hi Keyong,
>>
>> Instead of picking up a new shuffleId, can we reuse an existing shuffleId
>> after unregistering it? If the following plan worked, it would further
>> simplify the implementation:
>>
>> 1. Downstream tasks fail because of read failures.
>> 2. All active downstream tasks are killed, so the shuffleId is not used.
>> 3. An upstream vertex unregisters the shuffleId.
>> 4. The upstream vertex is re-executed normally. This re-execution
>> automatically registers the same shuffleId.
>>
>> In summary, we would like to go back in time before the upstream vertex
>> started by cleaning up the shuffleId. Could you please give a comment on
>> this plan?
>>
>> Thank you,
>>
>> --- Sungwoo
>>
>> On Sat, 30 Sep 2023, Keyong Zhou wrote:
>>
>>> Hi Sungwoo,
>>>
>>> I think your approach works with current architecture of Celeborn,
>>> and interpreting IOException when reading as read failure makes
>>> sense. Currently only when CommitFiles fails will LifecycleManager
>>> announce data lost.
>>>
>>> Thanks,
>>> Keyong Zhou
>>>
>>> Sungwoo Park <or...@pl.postech.ac.kr> 于2023年9月29日周五 22:05?道：
>>>
>>>>> Since the partition split has a good chance to contain data from almost
>>>> all
>>>>> upstream
>>>>> mapper tasks, the cost of re-computing all upstream tasks may have
>> little
>>>>> difference
>>>>> to re-computing the actual mapper tasks in most cases. Of course it's
>> not
>>>>> always true.
>>>>>
>>>>> To change from 'complete' to 'incomplete' also needs to refactor
>> Worker's
>>>>> logic, which
>>>>> currently assumes that the succeeded attempts will not be changed after
>>>>> final committing
>>>>> files.
>>>>>
>>>>>> a subset of succeeded attempts. In Erik's proposal, the whole upstream
>>>>>> stage will be rerun when data lost.
>>>>
>>>> Thank you for your response --- things are now much clearer.
>>>>
>>>> From your comments shown above, let me assume that:
>>>>
>>>> 1. The whole upstream stage is rerun in the case of read failure.
>>>>
>>>> 2. Currently it is not easy to change the state of a shuffleId from
>>>> 'complete' to 'incomplete'.
>>>>
>>>> Then, for Celeborn-MR3, I would like to experiment with the following
>>>> approach:
>>>>
>>>> 1. If read failures occur for shuffleId #1, we pick up a new shuffleId
>>>> #2.
>>>>
>>>> 2. The upstream stage (or Vertex in the case of MR3) re-executes all
>> tasks
>>>> again, but writes the output to shuffleId #2.
>>>>
>>>> 3. Tasks in the downstream stage re-try by reading from shuffleId #2.
>>>>
>>>> Do you think this approach makes sense under the current architecture of
>>>> Celeborn? If this approach is feasible, MR3 only needs to be notified of
>>>> read failures due to lost data by Celeborn ShuffleClient. Or, we could
>>>> just interpret IOException from Celeborn ShuffleClient as read failures,
>>>> in which case we can implement stage recompute without requiring any
>>>> extension of Celeborn. (However, it would be great if Celeborn
>>>> ShuffleClient could announce lost data explicitly.)
>>>>
>>>> An industrial user of Hive-MR3-Celeborn is trying hard to save disk
>> usage
>>>> on Celeborn workers (which use SSDs of limited capacity), so stage
>>>> recompute would be a great new feature to them.
>>>>
>>>> Thank you,
>>>>
>>>> --- Sungwoo
>>>>
>>>>
>>>>
>>>
>

Re: [PROPOSAL] Spark stage resubmission for shuffle fetch failure

[Keyong Zhou <zh...@apache.org>]

Hi Sungwoo,

Sorry for the late reply. Reusing a committed shuffleId does not work in
current architecture, even after calling unregisterShuffle in
LifecycleManager,
because the cleanup of metadata is delayed and not guaranteed. It will
be more complicated when we consider graceful restart of workers.

If we want to reuse the shuffleId, we need to redesign the whole picture.

Thanks,
Keyong Zhou

Sungwoo Park <or...@pl.postech.ac.kr> 于2023年10月2日周一 13:23写道：

> Hi Keyong,
>
> Instead of picking up a new shuffleId, can we reuse an existing shuffleId
> after unregistering it? If the following plan worked, it would further
> simplify the implementation:
>
> 1. Downstream tasks fail because of read failures.
> 2. All active downstream tasks are killed, so the shuffleId is not used.
> 3. An upstream vertex unregisters the shuffleId.
> 4. The upstream vertex is re-executed normally. This re-execution
> automatically registers the same shuffleId.
>
> In summary, we would like to go back in time before the upstream vertex
> started by cleaning up the shuffleId. Could you please give a comment on
> this plan?
>
> Thank you,
>
> --- Sungwoo
>
> On Sat, 30 Sep 2023, Keyong Zhou wrote:
>
> > Hi Sungwoo,
> >
> > I think your approach works with current architecture of Celeborn,
> > and interpreting IOException when reading as read failure makes
> > sense. Currently only when CommitFiles fails will LifecycleManager
> > announce data lost.
> >
> > Thanks,
> > Keyong Zhou
> >
> > Sungwoo Park <or...@pl.postech.ac.kr> 于2023年9月29日周五 22:05?道：
> >
> >>> Since the partition split has a good chance to contain data from almost
> >> all
> >>> upstream
> >>> mapper tasks, the cost of re-computing all upstream tasks may have
> little
> >>> difference
> >>> to re-computing the actual mapper tasks in most cases. Of course it's
> not
> >>> always true.
> >>>
> >>> To change from 'complete' to 'incomplete' also needs to refactor
> Worker's
> >>> logic, which
> >>> currently assumes that the succeeded attempts will not be changed after
> >>> final committing
> >>> files.
> >>>
> >>>> a subset of succeeded attempts. In Erik's proposal, the whole upstream
> >>>> stage will be rerun when data lost.
> >>
> >> Thank you for your response --- things are now much clearer.
> >>
> >> From your comments shown above, let me assume that:
> >>
> >> 1. The whole upstream stage is rerun in the case of read failure.
> >>
> >> 2. Currently it is not easy to change the state of a shuffleId from
> >> 'complete' to 'incomplete'.
> >>
> >> Then, for Celeborn-MR3, I would like to experiment with the following
> >> approach:
> >>
> >> 1. If read failures occur for shuffleId #1, we pick up a new shuffleId
> >> #2.
> >>
> >> 2. The upstream stage (or Vertex in the case of MR3) re-executes all
> tasks
> >> again, but writes the output to shuffleId #2.
> >>
> >> 3. Tasks in the downstream stage re-try by reading from shuffleId #2.
> >>
> >> Do you think this approach makes sense under the current architecture of
> >> Celeborn? If this approach is feasible, MR3 only needs to be notified of
> >> read failures due to lost data by Celeborn ShuffleClient. Or, we could
> >> just interpret IOException from Celeborn ShuffleClient as read failures,
> >> in which case we can implement stage recompute without requiring any
> >> extension of Celeborn. (However, it would be great if Celeborn
> >> ShuffleClient could announce lost data explicitly.)
> >>
> >> An industrial user of Hive-MR3-Celeborn is trying hard to save disk
> usage
> >> on Celeborn workers (which use SSDs of limited capacity), so stage
> >> recompute would be a great new feature to them.
> >>
> >> Thank you,
> >>
> >> --- Sungwoo
> >>
> >>
> >>
> >