Narrow and Wide Transformations
Narrow: each output partition depends on one input partition, so the work stays where it is. Wide: it depends on many, so the data must move. This single distinction predicts every stage boundary in your job.
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.
Which operations in this transformation are free, and which one just made the engine move the entire dataset across the network?
The reviewer of a pull request that adds one line to a transformation. Knowing which side of this line the new line falls on is the difference between a cost-neutral change and a doubled cluster bill.
The unit is the dependency between partitions: for one output partition, how many input partitions does it need? One means narrow. More than one means wide, and a shuffle (The Shuffle).
Treat all operations as roughly equivalent — a filter, a join, a distinct, a window are all just steps in a query. They read the same, they are written the same way, and the engine handles them.
A DISTINCT added defensively "in case there are duplicates" redistributes the whole dataset. The duplicates it was guarding against usually did not exist (Deduplication).
- A
DISTINCTadded defensively "in case there are duplicates" redistributes the whole dataset. The duplicates it was guarding against usually did not exist (Deduplication). - A window function partitioned by a different key than the aggregation adds a second full shuffle to a job that had one (Window Functions).
- A
repartitionadded to "increase parallelism" is a wide transformation and one of the more expensive things in the job, while acoalescereducing partitions is narrow and nearly free. - A filter placed after a join instead of before it means the join redistributes rows that are about to be thrown away — and the optimiser cannot move it if it sits behind a user-defined function (Predicate Pushdown).
What is actually happening
- A narrow dependency means each input partition feeds at most one output partition. The engine fuses every consecutive narrow operation into a single pass over the rows: filter, project and per-row expression all happen without materialising anything in between (Stages and Tasks).
- A wide dependency means one output partition draws from many input partitions. That requires redistribution, so it forces a stage boundary and a shuffle, whatever the operation is called.
- Recovery costs differ accordingly. Losing a narrow partition means recomputing one lineage chain from one input. Losing a wide one means recomputing from *all* of its parents, which is why lineage-based recovery gets expensive precisely where the job was already expensive (Partial Failure).
- The classification is a property of the dependency, not of the name.
joinis wide unless one side is broadcast, in which case it becomes narrow — the same operator, a different plan (Broadcast Joins). - A few operations sit awkwardly in between.
unionconcatenates partitions and moves nothing.coalescemerges neighbours without redistributing. Both are narrow despite changing the partition count (Partitions: the Unit of Parallelism).
The line, and which side each operation falls on
The rule is exactly one question: to produce one output partition, how many input partitions must be read? If the answer is one, the operation is narrow and the data stays put. If it is more than one, the data moves.
It is worth noticing what the table below does *not* organise by. It is not grouped by how the operation looks, how expensive its per-row work is, or how long it takes to write. A cheap-looking distinct() is wide; an expensive per-row regular expression over a huge column is narrow. The classification is about movement.
The two rows most often misread are repartition and coalesce. They sound like a pair and are not: one redistributes everything, the other merges neighbouring partitions in place. Reducing partition count with a repartition is a common and expensive mistake.
| Operation | Narrow or wide | Why | Recovery cost if a partition is lost |
|---|---|---|---|
filter, where | Narrow | Each row is judged alone; output partition n comes from input partition n. | Recompute one partition from one parent. |
select, per-row expressions | Narrow | A projection or a function applied within the row. | Recompute one partition from one parent. |
union | Narrow | Partitions are concatenated; no row changes machine. | Recompute one partition from its single source. |
coalesce (fewer partitions) | Narrow | Neighbouring partitions are merged without redistribution. | Recompute the merged group; still local. |
repartition | Wide | Rows are redistributed round-robin or by key across the network. | Recompute from every parent partition. |
groupBy / GROUP BY | Wide | All rows of a key must meet, and they start out spread across every partition. | Recompute from every parent that held that key. |
join (shuffle strategy) | Wide | Both sides are repartitioned by the join key so matching rows meet. | Recompute both sides' contributing partitions. |
join (broadcast strategy) | Narrow, for the large side | The small side is replicated everywhere; the large side never moves. | Recompute one large-side partition and re-broadcast. |
distinct | Wide | Duplicates can be in any partition, so keys must be regrouped. | Recompute from every parent. |
orderBy (global sort) | Wide | A total order requires range partitioning and a redistribution. | Recompute from every parent, including the sampling pass. |
Window with PARTITION BY | Wide | Each window's rows must be co-located before the function can be applied. | Recompute from every parent contributing to that window key. |
Order matters: the same operations, twice the cost
Because the shuffle moves bytes, everything that reduces bytes belongs before it and everything that expands them belongs after. That is the whole optimisation, and it is worth applying by hand in the cases the optimiser cannot reach.
The optimiser does this automatically for expressions it understands. It stops at the boundary of anything opaque: a user-defined function, a call into another language runtime, a predicate on the result of one. A filter it cannot see through is a filter it cannot move, and the shuffle then carries every row the filter would have removed (Query Optimizers).
The reverse mistake also exists. Pushing a filter through an outer join changes the result — rows that should have survived as unmatched nulls disappear — so the optimiser deliberately refuses. When a rewrite of that kind looks tempting, the question to ask is whether the two forms are actually equivalent, and the answer is often no.
Join the full order history to the customer dimension on `customer_id`, then filter to a single day and to non-cancelled rows. Both sides are shuffled in full; the day's worth of rows that survive is a tiny fraction of what crossed the network.
Filter orders to the day and to non-cancelled at the scan, project the four columns the join and the aggregate need, and only then join. The shuffle moves a small fraction of the rows and a fraction of the width of each one.
Shuffle cost is bytes moved, and the filter is what decides how many bytes exist to move. The optimiser performs exactly this rewrite when it can prove the two forms are equivalent — which it cannot do when the predicate hides inside a user-defined function, and must not do when pushing it below an outer join would change which rows survive (Predicate Pushdown).
Why the classification also decides recovery
The narrow/wide distinction is usually taught as a performance idea, and it is equally a fault-tolerance idea. The engine recovers a lost partition by recomputing it from its parents, so how many parents a partition has decides how expensive its loss is.
A narrow partition has one parent. Losing it costs one recomputation, on any executor, from data still available in storage. A wide partition has as many parents as there were input partitions — so recomputing it means re-running the whole producing stage, because the shuffle output it needed lived on machines that may also be gone.
This is the practical reason to materialise an intermediate result in long chains with several wide operations. Writing the post-shuffle result to storage converts an expensive lineage into a cheap restart point, at the cost of the write. For a job that runs nightly and must land, that is usually a good trade; for a job that runs in seconds, it is not (Checkpointing).
- 1Narrow chain
scan, filter, project, per-row expression — fused into one pass.
guarantees Partition count preserved; each output partition derives from exactly one input partition.
fails by Nothing structural. A lost task recomputes from the input files, cheaply, anywhere in the cluster.
- 2Wide boundary
Redistributes rows by key; materialises shuffle output on local disk.
guarantees Every row of a key lands together, for a fixed partition count.
fails by A lost executor destroying shuffle blocks, which forces the entire producing stage to run again (The Shuffle).
- 3Post-shuffle work
Aggregates or joins within each redistributed partition.
guarantees Correct per-key results, assuming the shuffle delivered every row of that key.
fails by Depending on many parents, so its recomputation is proportional to the whole previous stage rather than to one partition.
- 4Materialised checkpoint
Writes the post-shuffle result to durable storage before continuing.
guarantees A restart point that does not depend on any executor being alive.
fails by Costing a full write of the intermediate — worth it for long jobs, pure overhead for short ones.
The engine gives you recomputation for free and prices it by dependency count. Materialising is the manual override for the case where free recomputation is too expensive.
How to build it
Most important first.
- Do every narrow operation you can before the first wide one. Filters and projections applied earlier shrink what the shuffle has to move, which is the highest-leverage ordering decision available (Projection Pushdown).
- Count the wide operations in the plan and justify each one. A job with three shuffles on three different keys is usually two questions crammed into one query (Reading EXPLAIN ANALYZE).
- Arrange for wide operations to share a key where possible: aggregating and then joining on the same key can reuse one redistribution instead of paying two.
- Reach for
coalescerather thanrepartitionwhen you only need fewer partitions, and understand that coalesce reduces upstream parallelism because there is no shuffle to decouple it. - Express aggregates in a form the engine can combine partially. Sums, counts and bounded sketches reduce before the shuffle; exact distinct counts and medians cannot (Parallel Reduce).
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.
- Narrow transformations preserve partition count and, in practice, the arrangement of the data. Nothing about them promises anything regarding the *contents* of a partition.
- Wide transformations guarantee that all rows sharing a key end up together, and guarantee nothing about order or balance (Data Skew).
- Neither category guarantees anything about correctness. The classification is about data movement, and a wrong join is wrong at any width.
- A broadcast join guarantees the same result as a shuffle join only when the broadcast side genuinely fits — otherwise the job fails rather than silently returning a partial answer (Broadcast Joins).
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- Assert the row-count relationship each operation should have: a filter reduces, a projection preserves, an aggregation collapses to at most one row per key, a join multiplies by its fan-out. Wide operations are where these expectations are broken silently (Grain: What Does One Row Represent?).
- Track shuffle bytes per run. A change that turns a narrow plan into a wide one shows up here before it shows up as a missed schedule (Pipeline Metrics).
- Neither notices a join that is wide, balanced, fast and against the wrong key. The category says nothing about semantics.
- Narrow chains scale with cores and add essentially nothing to the wall clock beyond the work itself. Their duration is predictable.
- Every wide operation adds a barrier, so job duration grows in steps rather than smoothly as operations are added.
- Converting a wide operation to a narrow one — by broadcasting, or by reusing a partitioning — removes a barrier, which is usually the largest single improvement available to a job (Cost vs Freshness).
- Adding a column is narrow-neutral and widens everything a later shuffle has to move.
- Adding a
DISTINCT, a sort or a window function converts a one-shuffle job into a two-shuffle job — the plan changes shape from a one-line edit (Query Optimizers). - A join that used to be broadcast can silently become a shuffle join when the small side grows past the planner's threshold, turning a narrow operation wide with no code change at all (Broadcast Joins).
- A lost narrow partition is recomputed from its single parent, which is cheap and local.
- A lost wide partition needs every parent partition that fed it, which in the worst case means recomputing an entire upstream stage (Stages and Tasks).
- For long chains with several wide operations, materialising an intermediate result to storage converts an expensive lineage into a cheap restart point (Checkpointing).
What can go wrong
- An accidental wide operation — a defensive distinct, an unnecessary sort, a repartition believed to be a hint — doubling the job's cost.
- A filter that cannot be pushed below a join because it is expressed through an opaque function, so the shuffle carries rows destined to be discarded.
- A coalesce placed too early, which reduces parallelism for the entire upstream chain rather than just the write.
- A broadcast join that silently reverts to a shuffle join as data grows, changing the plan without changing the code.
- The mitigation failing: reordering operations to avoid a shuffle in a way that changes semantics — pushing a filter through an outer join alters the result, and the optimiser refuses to do it for exactly that reason.
- "Wide means slow, narrow means fast." Wide means data moves. A wide operation over a tiny pre-aggregated result is cheap; a narrow operation over a billion wide rows is not.
- "Repartition increases parallelism, so it is an optimisation." It is a full shuffle. It buys parallelism at the price of the most expensive operation in the system.
- "The optimiser will push my filter down anyway." It will, until the filter is expressed through a function it cannot see inside, at which point it silently will not (Query Optimizers).
- "A join is always wide." A broadcast join is narrow with respect to the large side, which is precisely why it is the most valuable rewrite available (Broadcast Joins).
Operating it
- The stage count in the physical plan, which equals the number of wide operations that survived optimisation (Reading EXPLAIN ANALYZE).
- Shuffle write bytes per stage, which attributes cost to the specific wide operation that caused it (Pipeline Metrics).
- The ratio of shuffle bytes to input bytes over time. A step change means the plan shape changed (Regression or Tuesday? Telling a Real Change from Noise).
- At 10x, narrow operations scale linearly with cores and wide ones scale with bytes moved. The gap between the two categories widens.
- At 100x, the number of wide operations effectively *is* the job's cost model, and eliminating one is worth more than any capacity decision.
- Higher key cardinality makes wide operations worse in a second way: more distinct keys means more partitions carrying state, and worse behaviour when the distribution is uneven (Partition Cardinality).
- Narrow operations cost CPU proportional to rows processed and nothing else.
- Wide operations cost serialization, local disk, network, merge and barrier time — the whole shuffle bill, once per wide operation (The Shuffle).
- The cheapest optimisation available in this domain remains the reordering of narrow operations before wide ones, because it costs nothing and reduces the largest term.
- Writing transformations to minimise wide operations makes them less readable. A three-shuffle query that expresses the question plainly may be the right choice for a job that runs monthly over small data.
- Broadcasting converts a wide operation into a narrow one and introduces a size assumption that will eventually be violated (Broadcast Joins).
- Bucketing eliminates shuffles for one specific join key and constrains the layout for every writer of that table (Bucketing).
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-SPECIFICThe narrow/wide vocabulary is Spark's. Trino and the distributed warehouses draw the same line and call the wide side an exchange; Flink calls it a keyBy or a rebalance. The classification is identical and only the words differ.
- GENERALEvery parallel data system distinguishes work that can stay local from work that requires regrouping, because it is a consequence of partitioned data rather than of any engine's design. Even a single-node engine pays a version of it as a cache-unfriendly repartition in memory.
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 recomputation from a dependency graph is a viable alternative to replication for intermediate state, and what it assumes about determinism to be correct.