cloud-fan commented on code in PR #57092:
URL: https://github.com/apache/spark/pull/57092#discussion_r3573938871
##########
PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md:
##########
@@ -0,0 +1,386 @@
+# Pipelined Shuffle Dependency & Concurrent Stage Scheduling
+
+A spec for running data-dependent stages of a single job concurrently,
connected by a shuffle the
+consumer reads incrementally.
+
+---
+
+## 1. Motivation
+
+Today, when two stages of a job are connected by a shuffle, they run one after
another: the consumer
+stage does not start until the producer's shuffle output is fully
materialized. (Independent stages —
+ones not connected by a shuffle — already run concurrently.) Some workloads
need even
+shuffle-connected stages to run **concurrently**, with the consumer reading
the producer's output
+**as it is produced** rather than after the producer finishes. This spec
introduces the scheduler
+primitives to express and run that.
+
+"Run these stages concurrently" and "the connecting shuffle is incremental"
are the same decision
+seen from two sides: co-scheduling a producer and consumer is only useful if
the edge is readable
+before the producer completes.
+
+---
+
+## 2. Primitives
+
+### 2.1 Pipelined shuffle dependency (PSD)
+
+A shuffle dependency declared **incrementally readable**: a consumer stage may
begin reading its
+output while the producer stage is still running.
+
+- It is a shuffle dependency (has a `shuffleId`, partitioner, map/reduce
sides); the *pipelined*
+ property is a binding part of the scheduler contract, not an advisory hint.
("Pipelined",
+ "incremental", and "transient" all describe this same edge in this doc —
read as its output being
+ streamed to the consumer as produced, not stored.)
+- The property is set during **physical planning** (an execution concern, not
a logical-plan one)
+ and carried into the `ShuffleDependency` the `DAGScheduler` reads at
stage-creation time.
+- It 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 one
+ job with both regular and pipelined groups uses the right implementation for
each. The scheduler
+ construct stays generic; the shuffle implementation stays pluggable.
+ - For example, an additional conf (e.g. `spark.shuffle.manager.incremental`)
can be introduced to
+ specify the incremental shuffle implementation used for pipelined shuffle
dependencies, alongside
+ the existing `spark.shuffle.manager` for the default.
+
+A **regular shuffle dependency (RSD)** is an ordinary shuffle dependency: its
output must be fully
+materialized before any consumer reads it.
+
+Note: the name *pipelined* is deliberately chosen over *streaming*. The
property is that a consumer
+reads producer output as it is produced — software-pipelining of dependent
stages — a general
+execution capability. Streaming / real-time mode (RTM) is the first caller,
but nothing about the
+primitive is streaming-specific.
+
+### 2.2 Pipelined group (PG)
+
+The set of stages connected to one another through pipelined edges — the
connected component of the
+stage DAG when only pipelined edges are considered.
+
+- A stage with no incident pipelined edge is a **singleton group** and behaves
exactly as a normal
Review Comment:
The group is the right *unit*, but I'd push on where it lives structurally —
the "singleton group" awkwardness is a symptom of a missing abstraction, not
just a naming choice.
Why it can't be a stage: a pipelined edge is a shuffle dependency, so it
splits producer/consumer into separate stages with different partition counts
(§3), and a `Stage` is one final-RDD / one homogeneous `TaskSet`
(all-`ShuffleMapTask` or all-`ResultTask`). So the co-scheduled unit is
irreducibly a container of >=2 member stages — not a stage, and not something
that can extend `Stage` (it would inherit the one-TaskSet contract it can't
honor).
The deeper issue: `Stage` today conflates two roles — (1) the
*task-submission* unit (one TaskSet) and (2) the *scheduling/lifecycle* unit
(what `waitingStages`/`runningStages`/`failedStages` track and what
`submitStage` acts on, `DAGScheduler.scala:164-170,1614`). A PG must be role
(2) but can't be role (1). "Singleton group" is what you get from papering over
the missing role-(2) abstraction — every stage is force-labeled a group, and
§4/§5/§6 then re-exempt singletons three times (this line; "gang admission
applies only to a group of two or more"; group-atomicity, vacuous for one
stage).
Concrete suggestion — factor the scheduling unit out of `Stage`:
```
sealed trait SchedulingUnit { def stages: Seq[Stage]; def externalParents:
Seq[Stage] }
case class SingleStageUnit(stage: Stage) extends SchedulingUnit //
ordinary stage
case class PipelinedGroup(members: Seq[Stage]) extends SchedulingUnit //
members.size >= 2
```
`Stage` is untouched (still the task-submission primitive). The changes are
localized: `submitStage` (`:1614`) and the three `HashSet[Stage]` (`:164-170`)
operate on `SchedulingUnit`; missing-parent readiness runs over
`unit.externalParents` (§4 group readiness); admission calls the existing
per-stage `submitMissingTasks` (`:1831`) once per member — gated by the gang
slot-check for a `PipelinedGroup`, unconditionally for a `SingleStageUnit`;
`submitWaitingChildStages` (`:1337`) fires on unit completion (§5
group-observable completion); failure/teardown act on `unit.stages` (§6
group-atomic). The size>=2 discrimination then happens once, structurally
(`case _: PipelinedGroup`), instead of being re-derived in each of §4/§5/§6.
There's a precedent for exactly this split one layer down: `TaskScheduler`
already separates the tasks (`TaskSet`) from their scheduling
(`TaskSetManager`). This is the same move at the `DAGScheduler` layer — `Stage`
= the tasks, `SchedulingUnit` = their scheduling. And it dissolves the
singleton problem for free: `SingleStageUnit` is the genuine one-member case of
a real abstraction (it submits one TaskSet), not a vacuous group-of-one.
Tradeoff to weigh: this refactors the scheduler's spine — `submitStage`, the
stage collections, `submitWaitingChildStages`, `abortStage`, and the UI's
stage-centric views all currently type on `Stage`, so it's a real (if
mechanical) change. The lower-footprint alternative is a side `Map[Stage,
GroupId]` plus the per-section branches §4/§5/§6 already imply — smaller diff,
but it keeps the discrimination scattered, which is the leak this finding is
about. Non-blocking, and squarely the kind of decision this discussion PR is
for.
##########
PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md:
##########
@@ -0,0 +1,386 @@
+# Pipelined Shuffle Dependency & Concurrent Stage Scheduling
+
+A spec for running data-dependent stages of a single job concurrently,
connected by a shuffle the
+consumer reads incrementally.
+
+---
+
+## 1. Motivation
+
+Today, when two stages of a job are connected by a shuffle, they run one after
another: the consumer
+stage does not start until the producer's shuffle output is fully
materialized. (Independent stages —
+ones not connected by a shuffle — already run concurrently.) Some workloads
need even
+shuffle-connected stages to run **concurrently**, with the consumer reading
the producer's output
+**as it is produced** rather than after the producer finishes. This spec
introduces the scheduler
+primitives to express and run that.
+
+"Run these stages concurrently" and "the connecting shuffle is incremental"
are the same decision
+seen from two sides: co-scheduling a producer and consumer is only useful if
the edge is readable
+before the producer completes.
+
+---
+
+## 2. Primitives
+
+### 2.1 Pipelined shuffle dependency (PSD)
+
+A shuffle dependency declared **incrementally readable**: a consumer stage may
begin reading its
+output while the producer stage is still running.
+
+- It is a shuffle dependency (has a `shuffleId`, partitioner, map/reduce
sides); the *pipelined*
Review Comment:
Modeling the pipelined edge as a `ShuffleDependency` subtype (as #57179
does) is the right call, but it cuts a line through `ShuffleDependency`'s
behavior that's worth stating explicitly in the spec, because the
implementation PRs have to get it exactly right. A pipelined edge *is* treated
as a real shuffle dependency for the two things you want inherited:
- **stage-boundary splitting** — `getShuffleDependenciesAndResourceProfiles`
matches `case shuffleDep: ShuffleDependency` (`DAGScheduler.scala:882`), so a
subtype splits the producer/consumer into separate stages automatically. This
is exactly what §3 ("introduces a shuffle boundary exactly as a regular one
does") relies on — without it there'd be no two stages to co-schedule.
- **transport** — `shuffleId`, `ShuffleWriter`/`Reader`, `ShuffleManager`
registration, all from the base ctor.
But it must *not* be treated as one for the post-boundary lifecycle:
`MapOutputTracker` registration (§6), `shuffleIdToMapStage` reuse (§4), and
executor-loss output-removal->resubmit (§6). The reason this is clean rather
than "inherit-then-disable" is that those three are all `DAGScheduler`-driven
at `createShuffleMapStage` (`DAGScheduler.scala:667-673`), gated per-dependency
— not forced by the type — so skipping them for a pipelined dependency costs
nothing at the dependency layer.
Suggest a short subsection making that split explicit (boundary + transport
= inherited; tracking + reuse + loss-resubmit = scheduler skips), since it's
the invariant every consumer of the marker type has to honor. Related: a marker
subtype relies on every existing `case _: ShuffleDependency` site meaning
"materialized" for the new type — the reuse/tracking sites are safe because
they're scheduler-gated as above; did you audit the remaining match sites, or
is that part of the follow-up? Non-blocking — the subtype approach itself is
sound.
--
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]
