Stages and Tasks
A stage is everything that can be done without moving data between machines. The boundary between two stages is always a shuffle, and it is always a barrier.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
Why does my job have five stages rather than one, and what decided where the cuts are?
The engineer reading a cluster UI at the wrong moment, and the schedule that depends on this job. Stage structure is the difference between a job you can reason about and a progress bar you can only watch.
A stage is a set of tasks that share the same shape of work. A task is one stage applied to one partition. Every number the engine reports — duration, bytes read, spill, retries — is reported at one of those two levels, and the interesting one is almost always the task.
Read the job as the sequence of operations you wrote: read, filter, group, aggregate, write. This is a fine mental model of the *result* and a poor model of the *execution*, because the engine does not run your steps in the order you typed them and does not run them one at a time.
The progress bar says stage 3 of 5 and nobody can say whether that means the job is 60% done. Stages are not equal, and the shuffle stage is usually most of the work (The Shuffle).
- The progress bar says stage 3 of 5 and nobody can say whether that means the job is 60% done. Stages are not equal, and the shuffle stage is usually most of the work (The Shuffle).
- A stage is at 999 of 1000 tasks for forty minutes. The stage is not nearly finished; it is entirely blocked on one task, and the next stage cannot start at all (Straggler Tasks).
- A filter written after a join appears in the plan before it, because the optimiser moved it. Reading the code as the execution order makes the plan incomprehensible (Query Optimizers).
- An executor is lost during stage 4 and the UI shows stage 3 running again. Its shuffle output vanished with the machine, so the stage that produced it must be recomputed (Partial Failure).
What is actually happening
- The driver builds a graph of transformations and cuts it wherever an operation needs rows from more than one input partition. Everything between two cuts becomes a stage, and everything inside a stage is pipelined — a row is filtered, projected and partially aggregated in one pass, without materialising each step (Narrow and Wide Transformations).
- Stage boundaries are barriers in practice. The next stage needs shuffle output from every task of the previous one, so it starts when the slowest of them finishes. That single fact explains most of the wall-clock behaviour of distributed jobs (The Critical Path Is the Only Path That Pays).
- Tasks within a stage are independent by construction: same code, different partition, no communication. This is what makes them retryable and what makes the stage scale with cores (Task Parallelism vs Data Parallelism).
- The engine pushes partial aggregation to the end of the stage *before* the shuffle when it can — computing per-partition sums and counts locally so that only one row per key per partition crosses the network. This is the difference between shuffling raw rows and shuffling an already-reduced summary (Parallel Reduce).
- Stages form a DAG rather than a chain. Two independent branches — one per side of a join — can run concurrently, and the join stage waits for both (DAG (Directed Acyclic Graph)).
One query, two stages, one shuffle
Take the most ordinary analytical job there is: read a day of orders, drop the cancelled ones, group by country, sum the amount, write the result. Four operations, and the engine will run them as two stages separated by exactly one redistribution.
Everything up to and including the *partial* aggregation happens where the data already is. Each task reads its own partition, filters it, computes a running sum and count per country for the rows it happens to hold, and writes those small per-country partial results to local disk, bucketed by which reducer will need them.
Then the network is used, once. Each reducing task fetches the partials for its own countries from every map task, adds them up, and writes its output. The rows that crossed the network were not orders — they were one partial per country per partition, which for eight countries and a thousand partitions is a very small amount of data. This is why the placement of the aggregation matters more than almost anything else in the job (Parallel Reduce).
1-- What you write2SELECT country, SUM(amount) AS revenue, COUNT(*) AS orders3FROM orders4WHERE dt = DATE '2026-08-25'5 AND status <> 'cancelled'6GROUP BY country;7 8-- Stage 1 (one task per input partition, no coordination)9-- read the row groups for dt=2026-08-25 only <- partition pruning10-- project amount, status, country only <- column pruning11-- filter status <> 'cancelled'12-- partial aggregate: per-partition SUM and COUNT per country13-- write those partials to local disk, bucketed by hash(country) % N14--15-- ===== SHUFFLE: the only network hop for data =====16--17-- Stage 2 (one task per shuffle partition)18-- fetch this task's bucket from every stage-1 task19-- final aggregate: add the partials per country20-- write output filesThe rows that cross the network are partial aggregates, not orders. Replace SUM with an exact COUNT(DISTINCT user_id) and that stops being true — a distinct count cannot be combined from partials, so the raw values have to be redistributed instead.
What a stage promises and how it fails
The two levels — stage and task — have different promises and different failure behaviour, and treating them as one thing is what makes distributed failures feel arbitrary. A task failing is routine and usually invisible. A stage failing is an event, and it can rewind work you thought was finished.
The subtle case is the third row below. Shuffle output is not durable: it sits on the local disk of the executor that produced it. When that executor is lost, the data the next stage needs is gone, so the engine must re-run the stage that produced it — after that stage had already been reported complete.
This is why aggressive use of pre-emptible capacity can make a job slower rather than cheaper, and why the external shuffle service exists: it decouples the lifetime of shuffle output from the lifetime of the executor that wrote it.
- 1Task
Applies one stage's work to one partition on one core slot.
guarantees Isolation: no communication with any other task, so it can be retried anywhere.
fails by Running far longer than its peers and holding the barrier open (Straggler Tasks).
- 2Stage
Runs every task of the same shape, then materialises shuffle output.
guarantees Completion only when every task has completed. No partial progress is exposed downstream.
fails by Retrying entirely once a task exhausts its attempts, discarding the work of the tasks that had succeeded.
- 3Shuffle output
Holds each map task's partitioned output on the local disk of the executor that produced it.
guarantees Availability for as long as the executor lives. It is not durable storage and it is not replicated.
fails by Disappearing with a lost executor, forcing recomputation of an already-completed stage.
- 4Job
Runs all stages required by one action, in dependency order.
guarantees The action's result, or nothing — the commit is where atomicity lives, if the sink supports it (Atomic Publish).
fails by Losing every stage's work when it fails late, which is the argument for committing in bounded ranges.
Read down the failure column: failure gets more expensive at every level, and the level that surprises people is the third — completed work that stops being available because a machine went away.
| Trigger | Symptom | Cause | Response |
|---|---|---|---|
| 999 of 1000 tasks done, no progress for 30 minutes. | A stage that appears nearly complete and is not moving. | One task has far more data than the others, or is on a degraded machine. The barrier waits for it. | Compare max and median task input bytes. Equal bytes points at the machine; unequal bytes points at the data (Data Skew). |
| A completed earlier stage re-enters the running state. | Stage 3 runs again while stage 4 is in progress. | An executor holding stage 3's shuffle output was lost, so its blocks must be regenerated. | Check executor loss events. If the cluster runs on pre-emptible capacity, this is the cost of that choice, and an external shuffle service is the mitigation. |
| Every task in a stage finishes in milliseconds. | Thousands of tasks, negligible each, and a stage whose duration is mostly dispatch. | Partition count is far larger than the data justifies, often inherited from a default after a selective filter. | Coalesce after the filter and before the write, so the task count matches the surviving data (Partitions: the Unit of Parallelism). |
| The plan shows five shuffles where you expected one. | Many stages, each moving substantial bytes. | Several independent wide operations — a distinct, a window, a join and a group-by on different keys. | Reorder so that operations sharing a key share a shuffle, and remove the ones that were defensive rather than required (Query Optimizers). |
How to build it
Most important first.
- Count the stages in the plan before running anything expensive. Each boundary is a shuffle you are paying for, and the useful question is whether every one of them is necessary (Reading EXPLAIN ANALYZE).
- Put filters and projections as early as the semantics allow, so the stage that ends in a shuffle carries the smallest possible rows into it. The optimiser does much of this and cannot do it through a user-defined function (Predicate Pushdown).
- Prefer aggregations that can partially combine before the shuffle over ones that must see every row of a key at once. Sums, counts, minimums and maximums combine; an exact distinct count and a median do not (Reduction Ordering: The Sum Changed When the Worker Count Did).
- Break genuinely enormous jobs into separately committed ranges rather than one long DAG. A twelve-stage job that fails at stage eleven has thrown away everything, while twelve daily runs each commit (Incremental Processing).
- Watch the task duration distribution within a stage rather than the stage duration. The stage duration is a consequence; the distribution is the cause (Tail Latency: Why p50 Being Fine Does Not Help).
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- Every task in a stage sees exactly one partition and communicates with nobody. That isolation is what makes retry safe and correct — for deterministic tasks.
- A stage completes only when all of its tasks complete. There is no partial credit and no early start for the next stage in the general case.
- Shuffle output written by a stage survives as long as the executor that wrote it does. It is not durable storage, and losing it means recomputing the stage that produced it (Checkpointing).
- Nothing guarantees that stages are comparable in size, so stage counts are not progress. Four stages remaining can be nine tenths of the runtime.
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- Assert on row counts across the stage boundary you care about most, usually the aggregation. A
GROUP BYthat emits more rows than it consumed is a grain error, not a performance one (Grain: What Does One Row Represent?). - Compare the job's stage structure between runs. A plan that silently changed — because statistics changed, or because the engine was upgraded — is a common cause of "the same job is suddenly slow" (Regression or Tuesday? Telling a Real Change from Noise).
- Neither check catches a correct plan over wrong data, which remains the default failure of this domain (The Pipeline Succeeded. The Data Is Wrong.).
- The end-to-end duration of a job is the sum of its stage durations, and each stage duration is the duration of its slowest task. This makes the job's freshness a function of the worst task in the worst stage rather than of average throughput.
- Adding stages adds barriers. A plan with fewer, larger stages usually finishes sooner than one with many small ones, even when the total work is identical.
- Recomputation after an executor loss extends the run in a way that is invisible in the plan, which is why a job's duration has a long tail that capacity planning tends to ignore (Capacity Planning: Traffic to Machines).
- Adding a wide operation adds a stage and a shuffle. A
DISTINCTadded to a query in the name of safety is a full redistribution of the data (The Shuffle). - Adding a column changes nothing structurally and widens every row crossing every barrier, so the cost moves without the plan shape changing.
- Engine upgrades change how stages are cut and when partial aggregation is applied. A job's stage count is not a stable contract, and code that depends on it — for progress reporting, say — will break (Pipeline SLOs).
- Task failure: retried in place, cheaply, usually invisibly.
- Stage failure after exhausting task retries: the whole stage re-runs, and any stage whose shuffle output was lost re-runs with it. Recovery therefore reaches further back than the failure point.
- Job failure: everything not committed is gone. The remedy is smaller committed units rather than more retries (Planning a Backfill).
What can go wrong
- One task in a stage running far longer than the rest, blocking the barrier and idling the cluster (Straggler Tasks).
- A fetch failure at the start of a stage, which cascades into recomputing the previous one.
- A plan with far more stages than necessary because the query performs several independent wide operations that could have been combined.
- An enormous single-stage job whose failure at the end costs the entire run.
- The mitigation failing: adding executors to a stage that is barrier-bound, which shortens nothing because the barrier is waiting on one task.
- "Stage 3 of 5 means 60% done." Stages are not equal units. The shuffle-heavy stage is usually most of the job.
- "999 of 1000 tasks complete means nearly finished." It means one task is deciding the stage's duration and the cluster is idle behind it.
- "Tasks in a stage talk to each other." They do not, by construction. All communication happens at the boundaries, which is exactly why the boundaries are expensive.
- "The execution order is the order I wrote." The optimiser reorders, fuses and pushes down. Read the plan, not the code (Lazy Evaluation).
Operating it
- The stage graph itself, with shuffle read and write bytes annotated on each boundary. It is the single most informative artefact a distributed job produces (Reading a Flame Graph).
- Task duration distribution per stage — median, 95th percentile and maximum on the same axis (Percentiles: Which One, and How Many Users Is That?).
- Stage retry counts, which distinguish "slow" from "repeatedly failing and eventually succeeding" (Pipeline Metrics).
- At 10x, stage structure is unchanged and each stage has more tasks. This is the well-behaved case distribution exists for.
- At 100x, barriers dominate and the shape of the plan matters more than the size of the cluster. Eliminating one shuffle is worth more than doubling capacity (Broadcast Joins).
- With very high task counts, scheduling throughput on the driver becomes its own limit — an effect that only appears at scale and looks like a mysteriously idle cluster.
- Each barrier costs the difference between the slowest task and the average one, multiplied by the whole cluster. That gap is the purest form of waste in distributed processing (Compute Waste).
- Each shuffle costs local disk writes, network transfer and reads. Stage count is therefore a direct cost driver, not just a structural detail.
- Recomputation after executor loss is paid twice, and it is a hidden cost of aggressive use of pre-emptible or spot capacity (Idle Capacity: Headroom or Waste?).
- Fewer, larger stages reduce barrier overhead and increase the cost of any single failure. More, smaller stages are the opposite trade, and neither is right in general.
- Pipelining within a stage is fast and makes intermediate results unavailable for inspection, so debugging a wrong number often means deliberately breaking the pipeline to materialise something.
- Partial aggregation before the shuffle is almost always right and changes floating-point results depending on grouping order, which matters for reconciliation to the last cent (Reduction Ordering: The Sum Changed When the Worker Count Did).
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- ENGINE-SPECIFICStage boundaries that materialise shuffle output to local disk are Spark's design. Trino streams between stages without materialising, so it has lower latency and no way to recover a failed worker by recomputation; Flink deploys operators continuously and checkpoints instead of staging.
- SIMPLIFIEDTreating a stage boundary as a hard barrier ignores engines that begin reducing as map output arrives, and adaptive execution that re-plans a stage from the statistics of the one before it. The barrier model predicts the observed behaviour of batch jobs well enough to reason with, and is not literally how every engine schedules.
- GENERALThe underlying rule — computation splits at the points where data must be redistributed — holds for every parallel data system including distributed warehouses, which simply do not show you the boundaries.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns why a barrier across many machines is expensive in the general case, and why recomputation from a dependency graph is a legitimate alternative to replicating intermediate state.