cloud-fan commented on code in PR #57092:
URL: https://github.com/apache/spark/pull/57092#discussion_r3584580721
##########
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:
Option 3 (dynamic recomputation) is the one I'd take — it's the best
resolution of this thread, and it's better than my original `SchedulingUnit`
proposal because it *removes* the awkward construct instead of wrapping it.
The thing worth being explicit about: option 3 changes how we should
*describe* this in the spec, because it retires `SchedulingUnit` as a logical
concept, not just as an implementation. `SchedulingUnit` existed to give the
lone stage and the group a common supertype so the size>=2 discrimination
happened once, structurally — but option 3's whole thesis is that `Stage` stays
the single scheduling primitive and group-ness is a *derived predicate over the
immutable stage graph* (`isPipelinedGroupMember`, `pipelinedGroupOf`,
recomputed on demand). There's no supertype to name and no `SingleStageUnit`
wrapper. Putting `SchedulingUnit` in the spec would just recreate the same
spec-vs-code friction from the other side.
So the clean framing isn't "abstract over the singleton" — it's "eliminate
it": a pipelined group has >=2 members by construction (it's a connected
component over edges, and an edge has two endpoints), and a stage with no
pipelined edge simply *isn't in a group* — it keeps the existing per-stage
behavior because it's a non-member, not because it's "exempted." The §4/§5/§6
re-exemptions then dissolve on their own.
Two spec edits follow from this and are worth doing in the same pass, since
they currently assert the *stored*-membership model option 3 replaces:
- §2.2: drop the "singleton group" construct — state the >=2 invariant and
that a non-member is an ordinary stage.
- §3: "Every stage belongs to exactly one group (singletons included). Group
membership is fixed at stage-creation time." — both halves go wrong under
option 3. A non-member belongs to *no* group, and membership isn't *stored* at
creation; it's *derived from the graph wherever a decision needs it*, stable
only because the graph is immutable. I'd reword to something like: "group
membership is a function of the pipelined-edge structure of the stage graph,
recomputed on demand; it is stable because that graph does not change after
creation."
One caveat on the "no persistent state, correct-by-construction" claim for
option 3: the `HashMap[Stage, DeferredCompletion]` buffer *is* persistent state
that still needs teardown on abort/job-cancel, so option 3 narrows the
lifecycle surface rather than eliminating it — the deferral buffer is the one
piece that can still go stale. Worth stating that explicitly so the "nothing to
keep consistent" framing isn't overclaimed. But group *extent* being derived is
a real win, and it's the right call.
--
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]
