jerrypeng commented on code in PR #57092:
URL: https://github.com/apache/spark/pull/57092#discussion_r3582599701
##########
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:
From an implementation perspective I think this is the right abstraction so
the we stops saying "the group, except when it's one stage, in which case…"
everywhere.
I traced the change against the code to size SchedulingUnit vs. the lighter
Map[Stage, GroupId], and want to lay out where the two actually differ, because
it's narrower than it first looks.
**What's identical in both.** The §5/§6 behavior — a member that finishes
early defers until the whole group is done; any member failure fails the whole
group — is the feature, and it exists regardless of representation. In both
approaches it's a single extracted funnel that the ~12 per-stage
`markStageAsFinished` / `submitWaitingChildStages` call sites
in`handleTaskCompletion` redirect to (e.g. `markStageComplete(stage)`), with
the "grouped? → all members done? → defer/finish" logic living once inside it.
The map version discriminates there with `groupIdOf.get(stage) match {
Some/None }`, the type version with `unitOf(stage) match {
PipelinedGroup/SingleStageUnit }` — same shape, one line. So the
completion/failure logic is *not* scattered or duplicated in either; both
funnel it.
**Where they genuinely differ: what the three scheduling collections are
keyed on.** `waitingStages`/`runningStages`/`failedStages` are referenced ~51
times across **19 methods** in `DAGScheduler` (plus `AppStatusListener` and
`DAGSchedulerSource`).
- `Map[Stage, GroupId]`: the collections stay `HashSet[Stage]`; those 19
methods keep thinking in stages and consult the map where they need group
semantics. ~150–250 LOC, contained to
`DAGScheduler.scala`, additive/low-risk. Cost: group-awareness is a lookup
you must remember at each collection-touching site.
- `SchedulingUnit`: re-type the collections to `HashSet[SchedulingUnit]`,
so those 19 methods (and the two other files) think in units and discrimination
is a compiler-checked `case`.
~400–600 LOC across 4 files. Cost: re-typing a hot, widely-referenced
field.
The UI and listener surface stays `Stage`-typed either way — I checked:
stage events are built from `stage.latestInfo`
(`SparkListenerStageSubmitted/Completed`, `DAGScheduler.scala:1911/1926/3481`)
and `AppStatusListener`/`ui/jobs/*` consume `StageInfo`/`stageId`, never a
scheduling unit — so a group's members just emit their own per-stage events as
today. `abortStage` likewise stays `Stage`-typed. So the observability spine
doesn't move in either approach; the only propagation is into the scheduling
collections, and that only under `SchedulingUnit`.
So the decision reduces to: **is compiler-enforced `case PipelinedGroup |
SingleStageUnit` across the 19 collection-touching methods worth ~250 extra LOC
and re-typing a hot field, versus a map lookup at those sites?
** Perhaps for M1 we can focus on getting single PG to work and **start with
`Map[Stage, GroupId]` ** (small, low-risk, and the completion funnel — the
correctness-critical part — is already extracted and identical) and **lift to
`SchedulingUnit` in M2**, when multiple concurrent groups make the per-site
lookups across those 19 methods actually painful and the compiler-enforced
destructuring earns its cost. That keeps the model you're describing as the
stated end-state without paying the spine re-type before the feature that
stresses it lands. What do you think?
--
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]
