[SPARK-58398][CORE] Group-atomic failure and fail-fast rejection for pipelined-shuffle stage groups ### What changes were proposed in this pull request? This PR combines the two failure-handling layers of the pipelined-shuffle work into a single reviewable change, on top of the now-merged concurrent scheduling of a `PipelinedShuffleDependency` group (SPARK-58263 / #57341): the **group-atomic failure model** and the **fail-fast rejection of unsupported idioms** that together make co-scheduling safe under failure. (It builds on two merged predecessors -- the `PipelinedShuffleDependency` definition and type-based routing from SPARK-58185 / #57286, and the concurrent scheduling from SPARK-58263 / #57341 -- and targets `master` directly; its net-new content is the failure and fail-fast layers described below.) A pipelined shuffle is transient and incrementally read: its output is streamed to a co-scheduled consumer and is never materialized as durable, re-readable map output. That has two consequences the scheduler must enforce. **Group-atomic failure (a pipelined group succeeds or fails as a whole).** A transient shuffle cannot be recomputed in isolation and its members run concurrently, so the stock "resubmit one stage" recovery does not apply -- any member failure must fail the whole group, which the caller then reruns (a streaming query restarts the batch). - **Member task failure.** A pipelined group member's `TaskSet` is tagged `isPipelined` and pinned to `maxTaskFailures = 1`, so the first task failure aborts the set instead of retrying in place, which routes to a whole-group abort. - **Executor loss.** Losing an executor running a member task is force-counted as a member failure (except benign `TaskKilled` / `TaskCommitDenied`), aborting the group. - **FetchFailed on a member.** Handled by a dedicated branch that aborts the whole group rather than resubmitting the map stage in isolation (a lone-stage resubmit of a transient shuffle would deadlock the group). The failed executor's *regular* outputs are still unregistered from the `MapOutputTracker`, exactly as the base handler does, for the benefit of other jobs. - **No transient-producer resubmit.** A pipelined `ShuffleMapStage` records its completed partitions locally and monotonically (never in the `MapOutputTracker`) and never flips back to unavailable, so losing an executor that held an already-consumed pipelined output cannot make the scheduler resubmit the producer -- which would otherwise hang the producer's streaming writer waiting on termination acks from reducers that already finished. The `TaskSetManager` "Resubmitted" re-enqueue loop likewise excludes pipelined sets. - **Cross-job / cross-time reuse rejected.** A transient shuffle has no retained output for a second job to read, so binding a pipelined producer stage to a second job fails fast. **Fail-fast on unsupported idioms (spec S9).** A pipelined group is rejected up front, before any stage is created (so a rejection leaves no partial scheduler state), when it uses an idiom v1 cannot support: a producer feeding **more than one consumer** (1:N fan-out needs multicast, deferred); a **barrier**, **statically-indeterminate**, **checksum-mismatch-retry**, or **push-merge** producer; a **reliable RDD checkpoint** anywhere in a member's within-stage chain (it reintroduces cross-time reuse of a transient edge); or **members with differing resource profiles** (the gang slot check compares one demand against one profile's capacity, so v1 requires a single-profile group). These throw a typed `PipelinedShuffleUnsupportedException` (carrying the `PIPELINED_SHUFFLE_UNSUPPORTED` error class), which `handleJobSubmitted` matches by type. Main changes: - `DAGScheduler.scala` -- the FetchFailed group-abort branch, the no-resubmit handling of a pipelined `ShuffleMapStage`, cross-job reuse rejection, and `checkPipelinedGroupsSupportedInRDDGraph` / `checkPipelinedProducerSupported` fail-fast (typed exception). - `TaskSetManager.scala` -- `maxTaskFailures = 1` and force-counted executor loss for a pipelined set; exclusion from the "Resubmitted" re-enqueue loop. - `ShuffleMapStage.scala` -- monotonic local availability for a pipelined shuffle. - `PIPELINED_SHUFFLE_UNSUPPORTED` and `PIPELINED_SHUFFLE_CROSS_JOB_REUSE` error conditions. ### Why are the changes needed? Co-scheduling a pipelined group (#57341) is only safe if failure is handled at the granularity of the whole group: because the shuffle is transient and once-through, the stock per-stage resubmit recovery would either deadlock the group or hang a producer's streaming writer. This PR makes any member failure fail the group atomically (so the caller reruns it) and rejects up front the idioms whose recovery/semantics are incompatible with a transient, concurrently-read shuffle -- turning what would be a hang or a silently-wrong schedule into a clear, immediate failure. The two layers are combined into one PR because they are inseparable in review: the fail-fast rejections define exactly which group shapes the failure model must then handle, and both are gated on the same `PipelinedShuffleDependency` type. ### Does this PR introduce _any_ user-facing change? No. All new behavior is gated on a job using a `PipelinedShuffleDependency`, which nothing constructs yet, so every existing job is scheduled and recovers exactly as before. The new error conditions (`PIPELINED_SHUFFLE_UNSUPPORTED`, `PIPELINED_SHUFFLE_CROSS_JOB_REUSE`) can only surface for a job that uses a pipelined dependency. ### How was this patch tested? New unit tests in `DAGSchedulerSuite` and `TaskSetManagerSuite` cover: - group-atomic failure: `maxTaskFailures = 1` aborting a member set on the first failure; executor loss force-counted (and benign `TaskKilled` / `TaskCommitDenied` not force-counted); a member FetchFailed aborting the whole group rather than resubmitting a single stage; - no transient-producer resubmit: a post-executor-loss straggler success not resubmitting the producer; a completed pipelined producer's availability surviving executor loss; the "Resubmitted" loop excluding a pipelined set (including the partial-producer-on-decommission case); - cross-job reuse rejected; a group-atomic rerun resetting per-partition commit authorization; - fail-fast idioms: fan-out, barrier / indeterminate / checksum / push-merge producer, a reliable checkpoint in a producer's or a consumer's chain (including downstream in the consumer stage), and a mixed-resource-profile group -- each rejected up front; and that regular-shuffle idioms are NOT rejected (inertness of the fail-fast for a job with no pipelined dependency). The full `DAGSchedulerSuite` and `TaskSetManagerSuite` pass. ### Was this patch authored or co-authored using generative AI tooling? Co-authored with Claude Code (Opus 4.8) Closes #57361 from jerrypeng/stack/pipelined-shuffle-pr6-failfast. Authored-by: Boyang Jerry Peng <jerry.peng@databricks.com> Signed-off-by: Wenchen Fan <wenchen@databricks.com>
Spark is a unified analytics engine for large-scale data processing. It provides high-level APIs in Scala, Java, Python, and R (Deprecated), and an optimized engine that supports general computation graphs for data analysis. It also supports a rich set of higher-level tools including Spark SQL for SQL and DataFrames, pandas API on Spark for pandas workloads, MLlib for machine learning, GraphX for graph processing, and Structured Streaming for stream processing.
You can find the latest Spark documentation, including a programming guide, on the project web page. This README file only contains basic setup instructions.
| Branch | Status |
|---|---|
| master | |
| branch-4.x | |
| branch-4.2 | |
| branch-4.1 | |
| branch-4.0 | |
| branch-3.5 | |
Spark is built using Apache Maven. To build Spark and its example programs, run:
./build/mvn -DskipTests clean package
(You do not need to do this if you downloaded a pre-built package.)
More detailed documentation is available from the project site, at “Building Spark”.
For general development tips, including info on developing Spark using an IDE, see “Useful Developer Tools”.
The easiest way to start using Spark is through the Scala shell:
./bin/spark-shell
Try the following command, which should return 1,000,000,000:
scala> spark.range(1000 * 1000 * 1000).count()
Alternatively, if you prefer Python, you can use the Python shell:
./bin/pyspark
And run the following command, which should also return 1,000,000,000:
>>> spark.range(1000 * 1000 * 1000).count()
Spark also comes with several sample programs in the examples directory. To run one of them, use ./bin/run-example <class> [params]. For example:
./bin/run-example SparkPi
will run the Pi example locally.
You can set the MASTER environment variable when running examples to submit examples to a cluster. This can be spark:// URL, “yarn” to run on YARN, and “local” to run locally with one thread, or “local[N]” to run locally with N threads. You can also use an abbreviated class name if the class is in the examples package. For instance:
MASTER=spark://host:7077 ./bin/run-example SparkPi
Many of the example programs print usage help if no params are given.
Testing first requires building Spark. Once Spark is built, tests can be run using:
./dev/run-tests
Please see the guidance on how to run tests for a module, or individual tests.
There is also a Kubernetes integration test, see resource-managers/kubernetes/integration-tests/README.md
Spark uses the Hadoop core library to talk to HDFS and other Hadoop-supported storage systems. Because the protocols have changed in different versions of Hadoop, you must build Spark against the same version that your cluster runs.
Please refer to the build documentation at “Specifying the Hadoop Version and Enabling YARN” for detailed guidance on building for a particular distribution of Hadoop, including building for particular Hive and Hive Thriftserver distributions.
Please refer to the Configuration Guide in the online documentation for an overview on how to configure Spark.
Please review the Contribution to Spark guide for information on how to get started contributing to the project.