jerrypeng commented on code in PR #57092: URL: https://github.com/apache/spark/pull/57092#discussion_r3546907659
########## PIPELINED_SHUFFLE_DEPENDENCY_SPEC.md: ########## @@ -0,0 +1,272 @@ +# 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 a multi-stage job runs one stage at a time: each shuffle is fully materialized before the next +stage starts. Some workloads need the stages of a single job to run **concurrently**, connected by a +shuffle whose consumer reads 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. +- 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 — selected + per-dependency, not per-job. 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 — which is a +general execution capability. Streaming / real-time mode is the first caller, but nothing about the +primitive is streaming-specific, so the dependency and the group are named for the capability, not +the caller. + +### 2.2 Pipelined group (G) + +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 + stage today. +- The group — not the edge or the individual stage — is the unit of **admission**, **slot + checking**, **completion**, and **failure**. + +**External input of G:** a regular shuffle dependency whose consumer is in G and whose producer is +not — i.e. a normal materialized parent of the group. + +--- + +## 3. Group formation + +- **Stage decomposition is unchanged.** A pipelined dependency introduces a shuffle boundary exactly + as a regular one does; the set of stages and their partitioning are identical. The pipelined + property changes only *when* stages run relative to one another, never *how the plan is cut into + stages*. +- **Group = connected component over pipelined edges.** As stages are created, two stages joined by a + pipelined edge are placed in the same group; the group is the transitive closure. +- **Every stage belongs to exactly one group** (singletons included). Group membership is fixed at + stage-creation time. + +--- + +## 4. Scheduling & admission + +- **A pipelined edge is non-sequencing.** The consumer of a pipelined dependency does not wait for + its producer to materialize. (A regular-dependency consumer still waits — the default behavior.) +- **Group readiness.** A group is ready to be admitted when every external input of the group (its + regular materialized parents) is available — the same precondition a normal stage has today, lifted + to the group. Pipelined parents inside the group impose no readiness precondition. +- **Gang admission (all-or-nothing).** A pipelined group is admitted only if the cluster can currently + run all tasks of all member stages **concurrently**; admission then submits every member stage at + once. There is no partial admission — a pipelined group is never left with some members running + while others wait on slots the running members occupy. + - *A non-pipelined (singleton) group is unaffected:* it is admitted exactly as a normal stage is + today — submitted when its parents are available, filling slots incrementally, with no all-at-once + requirement. Gang admission applies only to a group of two or more stages connected by pipelined + 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 Review Comment: I think I misunderstood your original comment about how maxNumConcurrentTasks works. I will correct this. -- 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]
