Spark Execution Plans, Stages, and Shuffles

In an illustrative scenario, a data engineer builds up a transformation — read a table, filter it, join it to customers, group by region — and every line returns instantly. Then a single count() takes twenty minutes and the cluster is saturated. Nothing was slow until something was, because the earlier lines had recorded the work and deferred computing its rows; what they did do — building and checking the plan, in some cases listing files or inferring a schema — was cheap by comparison. Deferring row computation lets Spark optimize the work as a whole.

The configuration names and defaults here apply to Spark 4.2.0.

Rows wait for an action; planning can happen earlier

Spark’s programming guide describes transformations as lazy: they record operations on a dataset rather than immediately computing their rows. An action requests the result and triggers the work it needs.

Lazy evaluation is not a convenience; it is what makes whole-query optimization possible. If each step executed as written, the engine could not combine two filters, drop a column nobody selects, or decide that a join should be done differently because one side turned out to be small. Deferring until an action arrives lets the optimizer see the whole query at once.

Two practical consequences follow, and the first needs a distinction. What is deferred is computing rows; building and validating the plan, and some preparation, is not. A name or type error — a column that does not exist — is an analysis failure and can be raised while you are still writing transformations, depending on the API path, and reading a source can list its files or infer a schema before any action. What typically waits for the action is anything that depends on the data itself: a cast that fails on one row, a division by zero, a file that turns out to be unreadable. Those surface at the action rather than at the line that caused them. And an action executes the part of the chain that has not been computed yet: without persistence, a chain of transformations is recomputed for each action that needs it, which is why the same “instant” code path can run three times in one script — though a shuffle output Spark still holds can be reused, so the recomputed range is set by what was cached and what survived, not by the chain’s length alone.

From code to a plan

What the action triggers is the rest of the compilation and then execution — optimization and physical planning, whatever analysis has already happened. Spark’s EXPLAIN statement exposes that compilation — the default output shows only the physical plan; EXPLAIN EXTENDED, or df.explain(mode="extended"), prints the four sections below, and EXPLAIN CODEGEN or the matching DataFrame mode shows generated code separately — and reading the sections in order is the fastest way to understand what the engine intends to do.

Parsed logical plan. The query’s structure, with names not yet resolved.

Analyzed logical plan. Names resolved against the catalog into typed objects — where “column not found” appears.

Optimized logical plan. The result of applying optimization rules: filters pushed down, projections trimmed, constants folded.

Physical plan. The operators that will actually run — scans, exchanges, join implementations.

The component doing this is Catalyst. The Spark SQL paper describes it as a highly extensible optimizer with rule sets for four phases — analysis, logical optimization, physical planning, and code generation — and states that “Catalyst supports both rule-based and cost-based optimization.” The same paper is candid about how narrow the cost-based part was: in the physical planning phase, several physical plans are generated and one is selected with a cost model, but “at the moment, cost-based optimization is only used to select join algorithms.” Those four phases are the compilation stages; they do not map one-to-one onto the four sections EXTENDED prints, and code generation has no section there. The paper’s description also dates from 2015: later versions added statistics-driven optimization behind settings of their own — a cost-based optimizer with join reordering, off by default and dependent on collected statistics — so in the version you run, the join choice depends on rules, statistics, the operators the join type supports, configuration, hints, and adaptive execution together.

This is the same layered idea as a database’s query execution plan, with one difference worth holding onto: the physical plan Spark shows you before the query runs is not necessarily the plan that runs, because of adaptive execution below.

From a plan to running work

Spark’s own glossary defines the execution units precisely. A job is “a parallel computation consisting of multiple tasks that gets spawned in response to a Spark action.” Each job “gets divided into smaller sets of tasks called stages that depend on each other.” A task is “a unit of work that will be sent to one executor.” The driver plans and schedules; executors run the tasks.

action            -> one or more jobs
shuffle boundary  -> a new stage
partition of a stage -> one task

Partitions are therefore the unit of parallelism, and their count decides how many tasks exist. The programming guide gives 2–4 partitions per CPU as a starting rule of thumb, with one task per partition. Too few partitions leaves cores idle; too many turns the run into scheduling overhead. See job, stage, and task for how these appear in the UI.

Why shuffles draw the stage boundaries

The reason a job splits into stages at all comes from the original RDD paper, which classified dependencies between datasets into two kinds: “narrow dependencies, where each partition of the parent RDD is used by at most one partition of the child RDD, wide dependencies, where multiple child partitions may depend on it.” A map creates a narrow dependency; a join generally creates a wide one.

The paper gives two reasons the distinction matters, and both are visible in any Spark UI. Narrow dependencies “allow for pipelined execution on one cluster node” — a filter after a map runs element by element without anything leaving the machine. Wide dependencies “require data from all parent partitions to be available and to be shuffled across the nodes.” That is the stage boundary: everything up to the exchange runs pipelined, then rows — or the partial aggregates that stand in for them — are redistributed to the placement the next operation needs, then the next stage begins. An exchange is needed when the current placement does not satisfy that need; a join whose inputs are already partitioned as required, or an aggregation whose local partial results shrink the data before it moves, shuffles less than the operator name suggests, and some shuffle reads are local.

The second reason is recovery. With narrow dependencies only lost partitions need recomputing, in parallel; with wide dependencies, the paper notes, “a single failed node might cause the loss of some partition from all the ancestors of an RDD, requiring a complete re-execution.”

And the shuffle itself is the expensive part of the query: it consumes disk access, serialization, and network transfer with map-side tasks spilling sorted data to disk and reduce-side tasks reading the blocks they need. The optimization instinct that follows is simple to state — reduce the number of shuffles, and reduce what each one has to move — measured from the plan’s exchanges and the stage’s shuffle bytes, since the number of wide dependencies bounds the shuffles a query could do rather than counting the ones it will.

How a join gets chosen

Strategy	How it works	Fits when
Broadcast hash join	The small side is sent to every executor; no shuffle of the large side	One side fits comfortably in executor memory, and the join type allows that side to be the build side — an equality join, and not, for example, a full outer join
Sort-merge join	Both sides are shuffled by key and merged in sorted order	Two large tables on an equality key, or any equality join where broadcasting is not allowed
Shuffled hash join	Both sides shuffled; a hash table is built on one side per partition	One side is small per partition after shuffling
Broadcast nested loop	Every row compared against every row of the broadcast side	Last resort — non-equality conditions

The automatic choice is governed by size estimates and constrained by what the join permits: spark.sql.autoBroadcastJoinThreshold defaults to 10485760 bytes — 10 MB — and setting it to -1 turns off that size-based selection, though an explicit hint can still broadcast and adaptive execution has a threshold of its own; spark.sql.broadcastTimeout defaults to 300 seconds. When statistics are missing or wrong, the estimate is wrong, and the plan is wrong with it — but a sort-merge join on a small table is not by itself proof of a bad estimate, since the join type, the keys, the configuration, or a hint can rule broadcasting out even with accurate statistics.

Hints take precedence over the estimate, and the documented priority when several are present is BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL — but a hint is a preference the planner honors when the strategy is applicable, not a guarantee that it will be used. Hints are a reasonable tool and a poor habit: each one fixes an assumption about data sizes and join conditions that has to be revisited when either changes, and the better first move is usually to fix the statistics or the layout.

What adaptive execution fixes while the query runs

Plans built from estimates are wrong in predictable ways, so Spark re-optimizes using runtime statistics. Adaptive Query Execution is enabled by default since Spark 3.2.0 (spark.sql.adaptive.enabled), and the documentation describes three families of change:

• Coalescing post-shuffle partitions. After a shuffle, small contiguous partitions are combined, which removes the classic problem of a fixed shuffle partition count producing thousands of tiny tasks. The advisory size defaults to 64 MB, and it is worth knowing that by default it is not what governs the result: with spark.sql.adaptive.coalescePartitions.parallelismFirst set to true — its default — Spark “ignores the target size … and only respect the minimum partition size” of 1 MB, “to maximize the parallelism.” The documentation is explicit that ignoring the target “is the default case.” Set parallelismFirst to false when you actually want partitions sized toward the advisory value.
• Switching join strategies. If runtime statistics show one side is small enough, a planned sort-merge join becomes a broadcast join; a separate setting allows conversion to a shuffled hash join.
• Splitting skewed partitions in joins. Enabled by default, with a partition treated as skewed when it exceeds both a factor of the median (default 5.0) and a size threshold (default 256 MB), and then split so that the work spreads across tasks.

What it does not do is worth stating as plainly. AQE re-optimizes at shuffle boundaries using statistics the run produces; it cannot undo a bad data layout, avoid reading files that pruning should have skipped, or rewrite every skewed operation: it splits skewed partitions in joins and, with a REBALANCE partitioning hint, can split skewed output partitions, but a single huge group in an aggregation still lands in one task and needs its own remedy — see skew. Nor does it remove the value of understanding the plan: it changes the plan, which means the physical plan printed before the run may differ from what the UI shows afterwards.

Reading a plan when something is slow

1. Read the exchanges, and check what kind each one is. The plan distinguishes them: a shuffle exchange redistributes rows — or partial aggregates — to a new partitioning, by hash, range, or round-robin, and ends a stage, while a broadcast exchange copies one side to every executor instead. They cost different things — redistribution moves whatever survives the map side once, broadcasting sends the small side many times and needs it to fit in executor memory — so counting them together tells you little. Three shuffle exchanges where one would do is usually a query structure problem rather than a cluster size problem.
2. Check which join implementation was chosen, and work out why in order: the actual sizes and the statistics, then the join type and keys, then configuration and hints, then the initial plan against the final one. A sort-merge join on a table you believed was small may mean the estimate disagrees with you — or that the join type could not use a broadcast, or that a setting or hint ruled it out.
3. Look at the scan node for pushed filters and the columns actually read. Work avoided at the scan never has to be optimized later.
4. Compare task durations within a stage. A stage whose median task is seconds and whose maximum is minutes is a signal to investigate, not yet a diagnosis. Skew is one cause; a slow host, a failing disk, GC pauses, spilling, waiting on shuffle fetches, and a retried task all produce the same shape. Read the per-task input and shuffle bytes, the record counts, and which executor ran it: uneven bytes point at data skew. Equal bytes are not yet evidence of a bad machine, since a user function or an amplifying join can make one partition far more expensive at the same input size; check output rows and GC and spill before concluding. Where the work is genuinely equal and one host is repeatedly slow, replacing it or enabling speculative execution is what helps. Speculation also gives you another observation — but not a verdict: a fast second attempt elsewhere points at the host or at a transient condition that has passed, while a second attempt that is equally slow may mean an expensive partition or a bottleneck both attempts share, such as a throttled store or a saturated path.
5. Then look at the runtime record. The event log holds what actually happened — the initial plan, the changes adaptive execution made, the final plan, and the stage and task times and bytes — and diagnosis needs them together: a join switched to broadcast mid-run may already have paid for a shuffle it then abandoned, which the final operator names alone will not show. That record exists only if event logging was enabled and retained before the run.

References

• Apache Spark 4.2.0 documentation: Cluster Mode Overview, RDD Programming Guide, EXPLAIN, Hints, Configuration: Runtime SQL Configuration, Performance Tuning, and Monitoring and Instrumentation
• Michael Armbrust et al., Spark SQL: Relational Data Processing in Spark, SIGMOD 2015
• Matei Zaharia et al., Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing, NSDI 2012