jerrypeng commented on code in PR #57092: URL: https://github.com/apache/spark/pull/57092#discussion_r3547086590
########## 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 + needs more slots than the cluster can offer, the submission fails fast, since the group could + never become fully co-resident. + - *What "available slots" is:* the cluster's concurrent-task capacity — the number of tasks it can + run at once — reusing the value the barrier slot check already uses (`sc.maxNumConcurrentTasks`), + not `spark.default.parallelism` (a default partition count, unrelated to how many tasks can run + concurrently). +- **Single resource profile per group (v1).** A resource profile is the executor/task resource + requirement (cores, memory, GPUs, ...) a stage runs under; the number of concurrent slots is + defined *per profile* (a cluster may run many concurrent CPU tasks but few GPU tasks). Comparing + one demand against one capacity is therefore only well-defined when all member stages of a group + share a single profile. v1 requires that and rejects a mixed-profile group (fail-fast, §9); + per-profile accounting — checking each profile's demand against that profile's own capacity — is a + follow-up, not needed for the streaming shapes whose members share the default profile. +- **Co-residency.** Once admitted, all member stages of a group are simultaneously running. +- **Single ownership.** A pipelined group belongs to exactly one job; a pipelined producer is not + shared across jobs. + +### 4.1 Cross-group admission (multiple groups / groups vs. regular jobs) + +A pipelined group holds its slots for its whole run, so admission is decided against currently-free +slots, and a group that does not fit is failed rather than queued. + +- **Capacity is free slots, not total.** The slot check counts only slots not occupied by running + tasks — running groups' and regular jobs' tasks are subtracted — so a group is admitted only if its + full demand fits in the slots free at admission time. + - *How free capacity is measured:* the cluster's concurrent-task capacity (the number of tasks it + can run at once — what an all-at-once gang admission must check against, see §4) minus the tasks + currently running. +- **No waiting queue, no partial reservation.** A group that doesn't fit fails its admission; it never + sits in a queue holding slots incrementally. This keeps the scheduler change minimal and cannot + deadlock (a group never occupies slots a sibling is blocked on), matching barrier, which also fails Review Comment: will update -- 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]
