[!DO NOT MERGE!][DOC] Pipelined shuffle dependency & concurrent stage scheduling spec discussion#57092
[!DO NOT MERGE!][DOC] Pipelined shuffle dependency & concurrent stage scheduling spec discussion#57092jerrypeng wants to merge 8 commits into
Conversation
…heduling spec Baseline of the design spec for native concurrent stage scheduling in the DAGScheduler, as reviewed on PR apache#56055. Subsequent commits layer reviewer feedback one point at a time for a clean per-round diff. Co-authored-by: Isaac
|
Reposting @cloud-fan feedback from #56055 (comment) The spec is the right shape and answers the direction well — PSD as a binding contract (§2.1), the group as the failure/admission unit (§2.2), and activation by membership (§8, which removes the prototype's half-activated state). A few gaps before it's the implementation contract:
Minor: (a) fail-fast admission → for streaming, crash-loops on restart; the barrier path retries BarrierJobSlotsNumberCheckFailed N times — say whether admission is retried or terminal. (b) "pipelined" over StreamingShuffleDependency is the better name (not streaming-specific) — worth a sentence on why. |
|
Reposting @tgravescs feedback from: #56055 (comment) Overall I really like the new idea of having a Pipelined Shuffle Dependency and support it directly in DagScheduler and define the contract. This opens it up to be generally useful in other scenarios. I'm still going through some of the details but here are a few initial comments I think we need to clarify some of the gang scheduling wording.
This is a group with multiple pipelined stages (pipelined group), not normal group with a single stage (ie how it runs today) Or is this specifically specified somewhere that we need to do Gang Scheduling? I think this opens up some possibilities in the scheduler but I'm not sure it also addresses being able to have multiple shuffle managers or multiple ways to do the shuffle. If you are mixing groups where one is regular and the other is a pipelined group, within the pipelined group you would want to use the StreamingShuffleManager but in the regular group you would use the regular shuffle manager. I think you can make a MultiShuffleManager like is in the streaming shuffle proposal but I think within that it needs to be able to dynamically change per group. It seems like the ShuffleDependency should have a way to indicate what type of shuffle is required. This might be getting more in the future but thinking about it here if we are defining contracts and new ShuffleDependencies. |
…mit, failure framing Layer the first round of reviewer feedback on PR apache#56055 onto the baseline spec: - Cross-group admission and starvation (new 4.1): capacity measured against free slots; a group that does not fit fails admission rather than queuing (no FIFO, no partial reservation); retry is the caller's decision, not the scheduler's. - Slot check: reuse maxNumConcurrentTasks (not defaultParallelism); require a single resource profile per group, reject mixed (fail-fast). - Completion: in-group result-stage side effects must be idempotent (the streaming at-least-once model); deferring commit noted as a future augmentation. - Failure framing: state it as a mechanism -- single-stage resubmit is disabled for group members, so any member failure fails the group -- rather than enumerating a FetchFailed path that does not exist on a streaming edge; note the coarser-grained recovery vs. base Spark. - Naming note: "pipelined" is deliberate, not "streaming"; the capability is general and streaming is only the first caller.
jerrypeng
force-pushed
the
pipelined-shuffle-spec
branch
from
a2c71eb to
177a57e
Compare
|
@cloud-fan thank you for your detail feedback. I will address them inline.
Capacity is measured against free slots (running groups' and regular jobs' tasks subtracted from maxNumConcurrentTasks). 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. I think this is good enough for now. In the future, we can consider if we want to augment the spec with wait / queuing capabilities
Switched the definition to the cluster's concurrent-task capacity — the value barrier's slot check uses (sc.maxNumConcurrentTasks(rp)) — explicitly not defaultParallelism(). On resource profiles: v1 requires a group to be single-profile and fail-fast rejects a mixed-profile group; per-profile accounting is a documented follow-up (the RTM shapes are single-profile).
I will clarify this. The current proposal simply requires an in-group result stage's side effects to be idempotent — the standard streaming model, where a batch is re-delivered on recovery and the sink absorbs it; a group rerun is the same re-delivery. Deferring the commit to group completion is noted as a future augmentation if a non-idempotent committer ever needs support.
I will clarify / refine this a bit more
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. Since shuffleIdToMapStage reuse is keyed by shuffleId, a new batch never matches — and so never binds to — a prior batch's stage. It's a consequence of existing Spark behavior rather than something the spec needs to add.
Clarified in the doc. |
|
@tgravescs thank you for your detail feedback. I will address them inline.
Yes let me clarify that point. Gang admission applies only to a pipelined (multi-stage) group, not to a normal single-stage stage. For a normal group the behavior is the same as today.
Great question! I did mention this in the original design sketch for this here: """ A shuffle with incrementalHint = true is served by the configured incremental manager; everything else by the default — keeping the scheduler construct generic while the shuffle implementation stays pluggable. """ |
… scoping, per-dependency shuffle manager - Gang admission (4): scope the all-or-nothing slot requirement to a pipelined group explicitly; add a sub-bullet stating a non-pipelined (singleton) group is unaffected and runs as a normal stage does today. Clarifies that gang scheduling applies only to a group of two or more stages connected by pipelined edges. - Pipelined shuffle dependency (2.1): note the property is also the per-dependency selector for the shuffle implementation -- the shuffle layer maps a pipelined dependency to an incremental ShuffleManager and everything else to the default, so a mixed job routes each group to the right implementation per-dependency, not per-job. Add a config example (spark.shuffle.manager.incremental). - Minor: drop the "distinguished in diagnostics" clause from the two-failure-reasons bullet (4.1). Co-authored-by: Isaac
|
@cloud-fan @tgravescs @mridulm @viirya lets use this PR to discuss and iterate on the concurrent scheduling spec. thank you! |
|
Thanks for the details @jerrypeng ! @cloud-fan and @tgravescs have excellent queries which covers a lot more than what I would have asked !
Also ... from responses above:
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.
nit: This needs to factor in the stage requirements (resource profile/cores per task) * num partitions, across all stages in the group.
I am concerned about the formulation - commit handling tends to be tricky. For the case when ResultStage is part of PG : why cant we not have similar behavior as what currently exists ?
These are implementation details of a specific usecase. |
cloud-fan
left a comment
cloud-fan
left a comment
There was a problem hiding this comment.
1 blocking, 1 non-blocking, 0 nits.
The spec faithfully formalizes the model this thread converged on, and its claims about the reused machinery are accurate. Two issues, both in the load-bearing §4 / §4.1 admission definition.
Correctness (2)
- PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md:95 (blocking): §4 and §4.1 give two different definitions of "available slots" (total vs. free), and
sc.maxNumConcurrentTasksreturns total, not free — see inline - PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md:126 (non-blocking): "matching barrier, which also fails its slot check rather than queuing" is inaccurate — barrier auto-retries N times before failing — see inline
Verification
Checked both against the tree. sc.maxNumConcurrentTasks(rp) -> CoarseGrainedSchedulerBackend.maxNumConcurrentTasks sums executor.totalCores over active executors with no subtraction of running tasks, so it is total capacity; barrier's checkBarrierStageWithNumSlots compares demand to that total. That confirms F2: §4.1's free-slots framing does not follow from the primitive §4 names. For F1, DAGScheduler's BarrierJobSlotsNumberCheckFailed handler re-posts JobSubmitted up to spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures (default 40) times before failing the job.
| edges. | ||
| - **Slot check.** The group's aggregate concurrent-task demand — the sum of `numTasks` over member | ||
| stages — is compared against the number of available slots in the cluster (a slot is one task's | ||
| worth of capacity, so this is the maximum number of tasks that can run at once). If the group |
There was a problem hiding this comment.
§4 and §4.1 define "available slots" two different ways, and the primitive named here delivers §4's, not §4.1's.
- §4 (this line): demand is compared against "the maximum number of tasks that can run at once", reusing
sc.maxNumConcurrentTasks— that is total capacity. - §4.1: "Capacity is free slots, not total ... counts only slots not occupied by running tasks ... minus the tasks currently running" — that is free capacity.
sc.maxNumConcurrentTasks(rp) resolves to CoarseGrainedSchedulerBackend.maxNumConcurrentTasks, which sums executor.totalCores over active executors (via calculateAvailableSlots) and does not subtract running tasks. So it is total, and barrier's checkBarrierStageWithNumSlots compares demand against that total.
That makes §4.1's cross-group deadlock-freedom argument (admit only if the group fits in currently-free slots) not follow from what §4 specifies: an implementer who "reuses the barrier slot check" as §4 says gets total-capacity admission, under which two groups can each pass the check yet not co-fit — the partial-co-residency gang admission is meant to forbid.
Suggest defining free capacity explicitly as maxNumConcurrentTasks(rp) minus the tasks currently running under rp (a new computation, not barrier's check verbatim), and reconciling §4's "maximum number of tasks that can run at once" wording with §4.1's "minus the tasks currently running" so the two sections agree on which value the slot check uses.
There was a problem hiding this comment.
I think I misunderstood your original comment about how maxNumConcurrentTasks works. I will correct this.
| currently running. | ||
| - **No waiting queue, no partial reservation.** A group that doesn't fit fails its admission; it never | ||
| sits in a queue holding slots incrementally. This keeps the scheduler change minimal and cannot | ||
| deadlock (a group never occupies slots a sibling is blocked on), matching barrier, which also fails |
There was a problem hiding this comment.
"matching barrier, which also fails its slot check rather than queuing" isn't accurate for barrier. On BarrierJobSlotsNumberCheckFailed, DAGScheduler re-posts JobSubmitted and retries the slot check up to spark.scheduler.barrier.maxConcurrentTasksCheck.maxFailures (default 40) times before failing the job — it does not simply fail rather than queue.
The fail-fast, no-scheduler-retry choice for a pipelined group is a fine v1 decision; it's just the "matches barrier" justification that's wrong (and it's the opposite of the barrier-retry point already raised in this thread). Suggest either dropping the comparison, or stating the contrast: unlike barrier, which auto-retries the slot check N times, a pipelined group's admission is terminal and retry is delegated to the caller's batch loop.
|
@cloud-fan thank you for your review and feedback! I have addressed your comments. PTAL. |
|
@mridulm thank you for taking the time to review this even when you are on vacation! Let me respond to your comments in line:
Yup, this is and should be supported, will clarify.
Yup, will explicitly mention.
Short answer, Yes. You're right that shuffle is 1:many in general. Worth separating where that fan-out actually comes from: a pipelined edge only ever originates from the physical-planning rule marking an exchange, and one ShuffleDependency gets more than one consumer only via exchange/stage reuse — ReuseExchangeAndSubquery (self-join, shared subquery/CTE, self-union) or submitMapStage. Branching alone doesn't do it: without the reuse rule, two identical subtrees plan to two separate ShuffleExchangeExec instances, each with its own shuffleDependency (it's a lazy val per node), i.e. two independent 1:1 shuffles. So "fan-out of a pipelined edge" ≡ "a pipelined dependency got reused/shared." That makes fan-out fundamentally incompatible with a transient incremental shuffle, not just unimplemented: reuse relies on a durable, independently-fetchable materialized output that N consumers each read on their own schedule. A transient shuffle is a live, once-through stream from a still-running producer — there's no stored copy to hand to a second consumer, and concurrent readers would need fan-out-aware backpressure that doesn't exist. Fan-out becomes well-defined only over a persistent/replayable medium (e.g. a Kafka-backed shuffle, where consumers replay from offsets) — which is the persistentHint axis the spec already lists as a non-goal and defers. So 1:many isn't off the table forever; it's gated on that persistent-shuffle capability, which isn't on the table now. On how we do it? Single-ownership and 1:1 are the same mechanism — a pipelined ShuffleDependency is never reused or shared. A group is rejected fail-fast if a pipelined producer has more than one consuming stage or its stage carries more than one jobId. That single rule delivers both no-fan-out and no-cross-job-sharing. This also answers your
What you called out make sense, though I think it is an optimization we may not need to do for v1. The fail fast mechanism should be good enough for now. In regards to using FetchFailed errors to signal this scenario. The streaming reader should reuse the FetchFailed channel as the signal to the scheduler, rather than the scheduler inferring failure from executor loss. Executor loss is the over-eager signal — it fires even when the producer had finished and its output was fully consumed, so it can't tell a harmful loss from a harmless one. The consumer can: it's the one reading, so it knows whether a producer's disappearance actually cost it data. Having the reader raise FetchFailed when (and only when) a read genuinely fails makes the consumer's real experience the trigger — a post-consumption producer loss produces no fetch failure and no teardown, which is exactly the precision you're after. The one adaptation: a FetchFailed on a pipelined edge must route to group failure, not the base single-stage mapper-resubmit (which is disabled for group members, and which is itself the source of the resubmit-then-hang problem). So we reuse the FetchFailed signal and its plumbing, but the handler recognizes the pipelined edge and reruns the group rather than recomputing the one mapper. Recovery stays group-atomic (the transient edge can't be re-fetched); only the detection moves from "driver infers from executor loss" to "consumer reports via FetchFailed." I will add a section in the spec to describe this as an optimization post-v1
Yes, you are right. That was mine intention but perhaps this bullet point added confusion. The intention was simply to indicate this scenario i.e. a regular shuffle being part of a PG is not possible. In reality, like you mentioned and what is mentioned in section 3, it will just split the DAG into multipe PGs. Though we should implement a check to make sure such a invariants hold.
Section 9 describes unsupported interactions with existing mechanism within a PG So for "Push-based shuffle merge" it simply means it cannot be used as an incremental shuffle in a PG. Of course it can be used as input or output for a PG just like a regular shuffle as you mentioned.
Yes, it is simply describing the incapability of these mechanism to exist within a PG. It should work as it should outside of a PG.
Enforced at stage/group creation: we walk each member stage's within-stage RDD chain — the narrow-dependency lineage, stopping at shuffle boundaries — and reject if any RDD has a non-NONE storage level (RDD.getStorageLevel). It reuses the DAGScheduler's existing within-stage traversal, so it's the same primitive the scheduler already uses, not new machinery. The shuffle-boundary stop is what makes the scope right: a cached complete input reached across a materialized shuffle (e.g. the static side of a stream-static join, or a broadcast) is outside the within-stage chain and correctly not flagged — only a cached RDD inside a member's own stage, which would freeze partial incremental output, is rejected. TBH, its is safe to use a cached/persisted RDD in a PG but not sure if it would ever be useful especially since structured streaming queries don't such caching. I guess if you want to reuse the results from an RDD part of the PG in another spark job? I will drop this from the incompatibility matrix for now.
So my proposal here in the spec at least for v1 is that any retries are left to the caller. There is no built-in retry or queuing mechanism. I think that is good enough for v1. We can always augment the spec later to include retries.
Will make that more clear.
Let me clarify this. There is actually no change to the existing output-commit path. The only genuinely new integration point is state cleanup: a member whose tasks all succeeded leaves a fully-populated authorizedCommitters array (the coordinator only clears a slot when the holder fails), so on group teardown we must drop each member's coordinator state — stageEnd / fresh stage ids on rerun — or the rerun's commits would be denied against the dead attempt's holders.
I think you are referring a response of mine to @cloud-fan question: "Cross-job stage reuse (§3/§4). Each micro-batch is a new job, but the base scheduler reuses shuffle-map stages via shuffleIdToMapStage. What forces a pipelined edge to create a fresh stage rather than binding batch N to batch N-1's cached stage?" We are talking about the structured streaming use case there. Though I think you question is more general about exchange / shuffle re-use? If so, that this not support in a PG since the assumption currently is that the data in an incremental shuffle is transient thus cannot be re-used. |
…ti-group PGs, fan-out ownership, commit path, invariants Addresses @mridulm's review comments on the spec: - PG outputs (S2.2, S5, S7): a group need not contain a ResultStage; it may have multiple materialized output edges feeding downstream groups, like a barrier stage embedded in a larger DAG. - Multiple concurrent PGs per DAG (S3): disjoint pipelined-edge sets form distinct groups that may run concurrently, subject to S4.1 slot arbitration. - Single ownership / 1:1 fan-out (S4, S9): enforced explicitly by never reusing a pipelined ShuffleDependency (bypass shuffleIdToMapStage), not by relying on streaming re-planning. Reuse must be prevented explicitly; from the scheduler's view a shuffle-map stage can be reused across jobs unless forbidden. - Output-commit (S5): the commit path is unchanged from base Spark -- OutputCommitCoordinator arbitration never contends inside a group (speculation rejected, single-stage resubmit disabled), so each partition has one attempt. The only integration point is clearing stale commit-coordinator state on group teardown. - Failure refinement (S6): post-v1, use a consumer-driven FetchFailed signal so a post-consumption producer loss does not force a rerun; on a pipelined edge it routes to group failure, not single-stage resubmit. - Internal regular shuffle (S7, S9): reclassified from fail-fast to an invariant that should hold by construction (S3) but is still checked defensively. - Push-based merge (S9): clarified it is rejected only as the incremental (pipelined) shuffle; a PG's regular input/output edges may still use it. - S9 scoping: all rejected idioms are scoped within a PG; a group's regular input/output edges are ordinary shuffles and unaffected. Co-authored-by: Isaac
…uestion on member StageCompleted timing Remove the open question asking whether SparkListenerStageCompleted for a member should be deferred to group commit or emitted in real time. The S5.1 table already specifies the behavior (deferred to group commit), so the trailing open-question paragraph was redundant. Co-authored-by: Isaac
…unchanged) A readability pass over the pipelined-shuffle spec. No semantic changes -- verified by sharded adversarial review (162 normative claims checked against the original; zero critical drift). Changes are wording, structure, and de-duplication only: - Unify the group abbreviation to PG throughout (was a mix of G and PG). - Fix a dangling reference in the S9 fan-out row: it pointed at a persistentHint axis in S2 that S2 never defined; reworded to 'a deferred capability v1 does not provide' (kind and reason unchanged). - Keep the free-vs-total slot-check rationale in one place (S4.1); drop the duplicated derivation from S4's 'How capacity is measured'. - Split the dense S5 output-commit bullet into two (idempotent commit + teardown state cleanup) and shorten the sentences. - Unify S5.1 terminology on 'group completion' (was 'group commit' / 'atomic commit' for the same milestone). - Regroup the S9 table rows to match the preamble's taxonomy (moot, incompatible, invariant); same rows, same kinds. - Add a short worked example in S2.2 (Z -> A -> B -> C) illustrating grouping, external input, concurrency, and materialized output. - Remove assorted restatements across S2.1, S2.2, S4, S4.1, S6, S7. Co-authored-by: Isaac
|
@cloud-fan @mridulm @tgravescs I have updated the spec incorporating your feedback Let me know if you have any more concerns. |
|
@cloud-fan @mridulm @viirya @tgravescs I also created the first PR that just declares PipelinedShuffleDependency. I think that part is no controversial? Pls take a look: |
|
Thanks for the response - the rest look good to me (I will review the full spec again later early next week) A few follow ups:
We should not assume PG/streaming shuffle is used only from sql/RTM - the constructs can be directly depended on - and I would expect some interesting usecases to develop as well.
Shuffle dependency is tied to the DAG, while stage is tied to a specific job execution of the dag.
This enforcement sounds good to me.
This would not work except for simple scenarios (linear chain of prefix* -> PG -> suffix*) - users would need to provision nontrivial resources to ensure there is sufficient capacity to run all stages which can run concurrently with a PG (including other PGs), and the PG itself. Given this is already support for barrier scheduling, it should be possible to adapt for PG as well ?
commit state is only for result stage ? And result stage is what determines job termination ? In other words, cleanup is triggered by result stage completion, and tied to commit -> which is what happens today as well ? |
|
@mridulm thank you for your feedback! Responding in line
Agreed on the framing, and working through it changed my answer on fan-out — for the better. Two things: On generality: you're right that PG / pipelined shuffle is a generic DAGScheduler construct and I shouldn't justify its properties via SQL/RTM specifics (ReuseExchangeAndSubquery, etc.). On fan-out — I'm walking back "incompatible." 1:N (one producer, many consumers) is a supported model; v1 just defers it. The mechanism differs from a regular shuffle, but the model is the same. Concretely, a consumer's PG membership picks its transport:
I would like to keep 1:1 mapping for now and defer 1:N later but like I mentioned it is definitely possible to extend it.
Agreed, it's worth stating precisely because the scheduler genuinely permits reuse. The root reason it must be prevented here is simple: a pipelined shuffle is transient — once-through, nothing retained — so unlike a regular shuffle there's no durable output for a second job to reuse. Reuse across jobs is therefore unsound for it, so we make run-once an enforced invariant rather than an assumption. I'll add a spec subsection making this explicit.
checkpoint is cache-adjacent but splits into two cases, and I'll handle them differently:
I will document it.
We could but can we defer it to v2? Initial callers such as RTM would not need such feature. I will mention it explicitly in the spec
The distinction I'm drawing is that a PG is an atomic scheduling unit — there are no per-task or per-partition replays within it. That matters because OutputCommitCoordinator is built on the opposite assumption: it arbitrates commits per task attempt, authorizing exactly one attempt per partition and denying any later request for a partition that has already committed. That's sound today because a committed partition is never recomputed — a stage rerun (e.g. on fetch failure) only recomputes the missing partitions, so a task that already committed never needs to commit again, and the permanent "deny" is exactly right. A PG breaks that assumption. Because the group is atomic, a failure anywhere reruns the whole group — including a result stage whose tasks already succeeded and committed. So committed So to let the rerun's tasks commit, we either (a) rerun the members under fresh stage ids — a fresh coordinator StageState with no prior committers — or (b) reset the committed state for those stages in the OutputCommitCoordinator (e.g. stageEnd on teardown) before the rerun. |
…t deferred, reuse layers, checkpoint, admission retry, commit reset Addresses @mridulm's follow-up comments on the spec: - Fan-out / generality (S4, S5, S9): reframe single-ownership at the construct level (the pipelined dependency and PG are generic DAGScheduler constructs, not SQL/RTM-specific). Split fan-out into two cases: cross-time / cross-job reuse stays incompatible (a transient once-through edge has no retained output to reuse), but concurrent branching (1:N co-scheduled consumers) is reclassified from incompatible to DEFERRED -- a supported model not built in v1 (needs multicast + per-consumer backpressure). A producer may also tee a durable materialized edge for out-of-group consumers; S5 gains deferred output registration so that dual-write is consistency-clean. - Cross-job reuse enforcement (S4): a pipelined shuffle is transient, so reuse across jobs is unsound and must be prevented at BOTH layers the scheduler would otherwise reuse a shuffle -- shuffleIdToMapStage stage reuse AND the MapOutputTracker-availability skip -- not just the former. - Checkpoint (S9): reliable RDD.checkpoint() in a member's within-stage chain is incompatible (durable lineage-truncated snapshot -> cross-time reuse; and a post-success recompute of a now-vanished transient input). Fail fast on checkpointData being a ReliableRDDCheckpointData. Cache / persist / local checkpoint remain whole-partition/ephemeral and allowed. - Admission retry (S4.1): v1 leaves retry to the caller, but that only suffices for a simple linear-chain PG run by a caller with its own restart loop. Name that limitation, and add scheduler-side PG-granular admission retry (mirroring barrier's transient-shortfall retry) as a post-v1 refinement. - Output-commit (S5): a PG is an atomic unit with no per-task replay, whereas OutputCommitCoordinator permanently denies re-commit of a committed partition (sound today because only missing partitions are recomputed). A group-atomic rerun reruns already-committed result tasks, so the rerun must reset per-partition commit authorization -- via fresh stage ids or stageEnd on teardown. Co-authored-by: Isaac
3 participants
PR created only to facilitate the discussion of introducing pipelined shuffle dependency & concurrent stage scheduling capabilities into the DAGScheduler that started here:
#56055