mridulm commented on PR #57092:
URL: https://github.com/apache/spark/pull/57092#issuecomment-4911827986
I am traveling/on vacation, so some quick thoughts/queries.
@cloud-fan and @tgravescs have excellent queries which covers a lot more
than what I would have asked !
A few quick notes:
* Can we call out that a PG can have multiple "output" (regular shuffle)
when `ResultStage` is not part of it ?
* Also, just as with barrier stage - a submitted DAG could have multiple
pipelined group within it - with these PG's potentially running concurrently.
* In general, shuffle is one to many (1 producer, many consumers) - are we
restricting/enforcing it to 1:1 ? If no - what is the behavior if/when it is
used across jobs (DAGs) ?
* "Single ownership." section has `a pipelined producer is not shared
across jobs` - how do we enforce this ?
* `Teardown is by group membership, not producer availability.`
* We do track shuffle loss when there is executor loss - should be
possible to plumb that into PG as well.
* Btw, this might be more strict than it needs to be ?
* For example, if all output from a producer has been consumed
successfully, and then the executor/node going down -> does not actually impact
the DAG ? would fetch failure be a better way to identify this failure mode ?
* It is fine to start with the stronger formulation btw !
* "Regular shuffle internal to a group — fail-fast / unsupported." -> I did
not understand why this needs to fail.
* we would simply split the DAG into two PG ?
* Section 9: "Push-based shuffle merge" -> can you clarify this why this
cant be supported for input and/or output ?
* Note - within the PG, this does not make sense anyway.
* Section 9 "Statically-indeterminate producer" and "Checksum-mismatch" -> I
did not understand this : a child of PG could have failed when reading PG
output, and so requires PG reexecution - and so might be impacted by
(INDETERMINATE) parent.
* Whether parent is indeterminate or not, the behavior would be same
though - agree if that is what was meant here.
* "Cached/persisted RDD in a member's within-stage chain" -> how we do
enforce this ?
Also ...
> A group that doesn't fit fails its admission attempt rather than queuing —
no waiting queue, no partial reservation, so it matches barrier and can't
deadlock on slots a sibling holds.
Just because at submission time there were insufficient resources to run it,
does not mean that will continue to be the case. See existing barrier mode for
insights.
I believe this is what Wenchen is also driving at.
> Slot check
nit: This needs to factor in the stage requirements (resource profile/cores
per task) * num partitions, across all stages in the group.
> Output-commit
I am concerned about the formulation - commit handling tends to be tricky.
Given a single result stage - why cant we not have similar behavior as what
currently exists ?
(I am assuming this only applies when result stage is part of the PG - not
side channel writes)
> This can't happen across micro-batches: each micro-batch is a new job
built from a fresh query plan (a new IncrementalExecution), whose exchanges
mint new ShuffleDependency objects with new shuffleIds.
These are implementation details of a specific usecase.
From scheduler perspective, you can have reuse - unless we explicitly dont
track streaming shuffle ids (which is an option).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
