jerrypeng commented on code in PR #57092:
URL: https://github.com/apache/spark/pull/57092#discussion_r3583906608
##########
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:
Btw another potentially less intrusive way to implement this is just compute
group membership dynamically. There is no GroupId, no Map[Stage, GroupId], and
no SchedulingUnit. Instead, Stage remains the single scheduler primitive, the
three lifecycle collections (waitingStages/runningStages/failedStages) stay
HashSet[Stage] untouched, and group identity is derived on demand from the
RDD/stage graph wherever a decision needs it — pipelinedGroupOf walks the
pipelined edges to assemble the member set for the slot check,
isPipelinedProducer/isPipelinedGroupMember are per-stage predicates over the
same graph, and the group's §5/§6 behaviors are attached to individual stages
(the deferral buffer keyed as HashMap[Stage, DeferredCompletion], replayed
through the existing event loop; group-atomic failure routed via
TaskSet.isPipelined and a per-stage membership check on the FetchFailed path).
Its advantages are that it adds no new persistent state and therefore no
lifecycle to keep con
sistent — nothing to populate at group formation or tear down on job abort,
stage retry, or eviction, so it categorically cannot go stale the way a side
map or unit index can; because group extent is always recomputed from the
authoritative graph, it is correct-by-construction across retries and aborts.
It is also minimal and additive: it re-types no hot, widely-referenced field
(the ~57 references to the scheduling collections and the entire Stage-typed
observability/task-submission surface are left exactly as they are), so the
change is contained, low-risk, and easy to review, and it keeps Stage as the
one task-submission primitive rather than introducing a parallel scheduling
type that the UI, listeners, and abortStage would otherwise have to learn about.
--
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]
