How Spark Decides What to Run: Plans, Stages, and the Cost of Moving Rows

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, and the reason is that none of the earlier lines did any work. The example is invented; the behavior is by design, and understanding it is most of what separates tuning Spark from guessing at it.

This article follows one query from code to running tasks: how a plan is built and optimized, how it becomes stages and tasks, why shuffles are the boundaries that matter, how joins are chosen, and what adaptive execution changes while the query runs. Configuration names and defaults are from Spark 4.2.0 documentation, checked in September 2026; they change between versions.

Nothing runs until something asks for an answer

Spark’s programming guide is explicit: “All transformations in Spark are lazy, in that they do not compute their results right away. Instead, they just remember the transformations applied to some base dataset… The transformations are only computed when an action requires a result to be returned to the driver program.”

Lazy evaluation is not a convenience; it is what makes optimization possible. If each step executed as written, the engine could never 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 is not always deferred. 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. 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 a chain of transformations is recomputed each time an action needs it unless the result is persisted, which is why the same “instant” code path can run three times in one script.

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 the stages of that compilation, and reading them in order is the fastest way to understand what the engine intends to do.

Plan	What it is
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.” Rules do most of the work; statistics decide the joins.

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 states the rule of thumb: “Spark will run one task for each partition of the cluster. Typically you want 2-4 partitions for each CPU in your cluster.” 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 are redistributed, then the next stage begins.

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: the programming guide calls it “an expensive operation since it involves disk I/O, data serialization, and network I/O,” 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.

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
Sort-merge join	Both sides are shuffled by key and merged in sorted order	Two large tables
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: spark.sql.autoBroadcastJoinThreshold defaults to 10485760 bytes — 10 MB — and broadcasting can be disabled by setting it to -1, with spark.sql.broadcastTimeout defaulting to 300 seconds. When statistics are missing or wrong, the estimate is wrong, and the plan is wrong with it.

Hints override the estimate. The documented priority is BROADCAST > MERGE > SHUFFLE_HASH > SHUFFLE_REPLICATE_NL, so a broadcast hint wins when several are present. Hints are a reasonable tool and a poor habit: each one is a hard-coded assumption about data sizes that will eventually stop being true, 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 fix skew that arrives outside a join — a single huge group in an aggregation, for instance. 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 every row by key and ends a stage, while a broadcast exchange copies one side to every executor instead. They cost different things — redistribution moves all the data 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 whether it matches what you expect of the data sizes. A sort-merge join on a table you believed was small means the estimate disagrees with you.
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, including the plan AQE settled on, which is the only version that matters for diagnosis.

References

Spark 4.2.0 documentation was checked in September 2026. Defaults and adaptive-execution behavior differ between versions; confirm them for the version you run.

• Apache Spark 4.2.0 documentation: Cluster Mode Overview, RDD Programming Guide, EXPLAIN, and Performance Tuning
• 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