The Shuffle
The one operation in a distributed job that uses the network for data. Every row is assigned a destination by key, written to local disk, fetched across the cluster and merged — which is why it dominates the runtime, the cost and the failure modes of almost every 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.
Why is the GROUP BY the expensive part of my job when the filter above it reads ten times as many rows?
Every downstream model that depends on this job landing on time, and the person who has to explain why the cluster bill doubled after a query changed by one line. The shuffle is where an innocuous edit becomes an expensive job.
A shuffle moves one row at a time, addressed by key. The unit that matters is therefore the (key, row) pair: how many distinct keys there are decides how well the work spreads, and how wide each row is decides how many bytes cross the network per key.
Write the join and the group-by that express the question, and let the engine work out how to execute them. It will, correctly. For most jobs this is the right level to work at, and premature shuffle-avoidance produces unreadable transformations that were never slow.
A join key that is fine in the source is catastrophic in the shuffle: nulls, empty strings and a default -1 all hash to one place, so a single reducing task receives every unmatched row in the dataset (Data Skew).
- A join key that is fine in the source is catastrophic in the shuffle: nulls, empty strings and a default
-1all hash to one place, so a single reducing task receives every unmatched row in the dataset (Data Skew). - A
SELECT *upstream means every column of every row crosses the network, including the two wide text columns nobody in the aggregate uses (Projection Pushdown). - The job runs for years and then an executor is pre-empted mid-stage. Its shuffle output is on its local disk, so the previous stage recomputes — and the run that normally takes twenty minutes takes an hour (Partial Failure).
- Local disk fills with shuffle spill during a wide aggregation and tasks start failing with no-space errors, on a cluster with plenty of memory and plenty of object storage (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
- The number of shuffle partitions was tuned once. The data grew, each reducing task now holds more than it can, and the stage becomes a disk-bound merge rather than an in-memory aggregation.
What is actually happening
- On the map side, each task computes a destination for every row — typically
hash(key) % partitionCount— and buffers rows per destination. When a buffer fills, it is sorted and spilled to local disk. At the end of the task the spills are merged into one file per task, with an index that says where each destination's bytes begin (Merge Sort). - On the reduce side, each task fetches its own slice from every map task's output file. That is M x R fetch operations for M map tasks and R reduce tasks, and the metadata alone becomes significant when both numbers are large (Network Signals: Is It the Network, or the Service on the Other End?).
- Everything crossing that boundary is serialized on the way out and deserialized on the way in. Serialization is often a larger share of shuffle cost than the network transfer itself, and it is the part that scales with row width rather than row count (Busy Is Not the Same as Working).
- Reducing tasks usually need their input grouped by key, which means sorting or hashing the fetched data. When it does not fit in memory it spills again, so a single logical redistribution can write the same rows to local disk twice (Memory Pressure, Swap and the OOM Killer).
- The whole thing is a barrier: no reducing task can produce a correct result until it has fetched from every map task, so the stage is gated by the slowest map task in the cluster (Stages and Tasks).
- The partitioner is the reason the shuffle works at all: because the destination is a deterministic function of the key, every row of a key ends up in the same place, and grouping becomes a local problem again (Hash Table).
What physically happens between two stages
The word "shuffle" suggests rows being handed around. What actually happens is closer to a distributed sort with a mailbox for every destination, and it is worth walking the mechanism in physical terms because every cost driver falls out of it.
Each map task processes its partition and, for each output row, computes a destination partition from the key. Rows accumulate in per-destination buffers in memory. When memory pressure builds, a buffer is sorted and spilled to local disk. At the end of the task all the spills are merged into a single file, plus a small index recording where each destination's region starts. This is the map side, and it is entirely local: no network yet.
Then the reducing stage starts. Reduce task 7 needs region 7 from every single map task, so it issues a fetch to each of them and streams the results in. Those bytes are deserialized, then grouped — by hash table or by merge — and if the result does not fit in memory it spills again. Only when a reduce task has fetched from every map task can it emit a correct result for any key, because until then it cannot know it has seen all the rows.
Count the trips a single row makes: serialized once, written to local disk once (at least), read back, sent over the network, deserialized, possibly written and read again during the reduce-side merge. That is why a shuffled byte is worth several scanned bytes, and why the useful question is never "how fast is the network" but "how do we move fewer, narrower rows".
- The destination is a pure function of the key, which is what makes grouping local again on the other side (Hash Table).
- The map side never touches the network; the reduce side never touches the input files. They fail for entirely different reasons.
- A reduce task cannot emit anything until it has fetched from every map task, which is what makes the boundary a barrier (Stages and Tasks).
Why it is expensive, in order
Engineers usually name the network first. The network is real, and in most jobs it is not the largest term. Serialization burns CPU proportional to row width; local disk absorbs every spill; the sort or hash on the reduce side burns CPU again; and the barrier converts the slowest task into everybody's wait.
The ordering below is the transferable part. The absolute magnitudes depend on the row width, the codec, the instance type and the cluster topology, and any number quoted for them is a benchmark for a machine you do not have (Benchmark Fallacies: Confident Numbers That Are Wrong).
What matters practically is that the top three drivers are all reduced by the same two actions: move fewer rows, and make each row narrower. That is why projection and pre-aggregation are worth more than tuning, and why they are applied before the shuffle rather than after it.
CPU proportional to row width and to the number of fields. This is the term that a SELECT * inflates without changing the row count at all.
Every byte that does not fit in the map-side buffer is written and read back — and possibly again on the reduce side (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
Grows with bytes moved and with cross-zone placement. Real, and usually not the leader.
Grouping the fetched rows by key. Cheap when the state fits in memory, and superlinear the moment it does not.
The gap between the slowest map task and the median, multiplied by every idle executor. Grows directly with skew (Straggler Tasks).
Negligible at hundreds of tasks, a first-class problem at tens of thousands on both sides.
Relative weights, not measurements. The ordering is what transfers between platforms; the magnitudes do not, and this domain never publishes a price.
Relative weights for a typical sort-based shuffle in a batch job. They exist to fix an ordering, not to predict a runtime — and the ordering is what tells you that narrowing rows beats every configuration change available.
Read every column, join first, then group by country at the end. Every raw row — including the wide `url` and `referrer` text columns — is serialized, spilled, fetched and merged, and the reduce side holds far more state than the result needs.
Project the three columns the aggregate needs, apply the date and status filters at the scan, compute per-partition partial sums and counts per country, and shuffle those. One small row per country per partition crosses the network, and the reduce side merges a few thousand tiny records.
Shuffle cost is bytes moved, and bytes are rows times width. Partial aggregation collapses the row count by the ratio of rows to distinct keys, and projection cuts the width — the two multiply. Neither changes the result, and both are unavailable to the optimiser if the aggregation is hidden inside a user-defined function it cannot see through (Query Optimizers).
The four ways to move less
There are exactly four strategies, and they are worth knowing as a list because the instinct — change a configuration value — is not among them. Reduce what crosses; combine before crossing; replicate instead of crossing; or arrange for the data to already be where it needs to be.
The fourth deserves particular attention, because it is the one that requires planning rather than a code change. If a table is physically bucketed on the key it is usually joined on, and the other side is too, the engine can pair up buckets directly and skip the redistribution entirely. That benefit is real, and it is limited to joins on exactly that key with a matching bucket count (Bucketing).
None of these removes the need to redistribute when a question genuinely requires global regrouping. A global sort, an exact distinct count across the whole dataset, or a join between two large tables on a high-cardinality key all require the data to move. The goal is to move it once, narrow, and with the work already partly done.
Which property of this job can you change?
when The shuffled rows carry columns the downstream operation never reads, or rows a later filter discards.
cost Almost nothing, and the optimiser usually does it for you — unless a user-defined function blocks it (Predicate Pushdown).
when The operation is a combinable aggregate — sum, count, min, max, approximate distinct — over many rows per key.
cost Map-side memory for the partial state, and a change in summation order that can move a floating-point total in the last digits (Reduction Ordering: The Sum Changed When the Worker Count Did).
when One side of a join fits comfortably in each executor's memory after filtering.
cost The small side is sent to every executor, so cost grows with executor count — and a wrong size estimate takes down the driver or the executors (Broadcast Joins).
when Both sides are already bucketed or sorted on the join key with compatible bucket counts, or a second aggregation uses the same key as the first.
cost A layout commitment that must be maintained by every writer, and which helps only the operations using exactly that key (Bucketing).
when The question genuinely requires a global regrouping of large data on a high-cardinality key.
cost Choosing a shuffle partition count that matches the volume, provisioning local disk for the spill, and revisiting both as the data grows (Partitions: the Unit of Parallelism).
1-- Two wide operations on two different keys: two shuffles.2SELECT c.country,3 COUNT(DISTINCT o.user_id) AS buyers, -- shuffle on user_id4 SUM(o.amount) AS revenue -- grouped by country5FROM orders o6JOIN customers c ON c.id = o.customer_id -- shuffle on customer_id7WHERE o.dt = DATE '2026-08-25'8GROUP BY c.country;9 10-- Same answer, one redistribution of the large side:11-- * customers is small after filtering -> broadcast it, no shuffle on the join12-- * exact COUNT(DISTINCT) still needs user_id rows to be regrouped;13-- an approximate distinct sketch combines from partials and does not14-- The point is not the rewrite. It is that each wide operation is a15-- redistribution, and the plan tells you how many you are paying for.Read the plan rather than the query to count shuffles: the optimiser may fuse two operations that share a key, and it will not fuse two that do not. An exact COUNT(DISTINCT) is the aggregate that most often forces an extra one (Reading EXPLAIN ANALYZE).
How shuffles fail, and what each failure looks like
Shuffle failures are distinctive because they are rarely about correctness and almost always about a resource that is not the one people check. Memory is fine, object storage is fine, CPU is idle — and the job is failing on local disk, or on fetch metadata, or on one key.
The most confusing of them is recomputation. A stage that has been reported complete for twenty minutes can start running again, because its output lived on the local disk of an executor that has since been taken away. Nothing is wrong; the engine is doing exactly what its recovery model says. It just makes the progress bar dishonest.
The last row of the table is the one worth internalising: the standard mitigation for one shuffle failure is the cause of another. Raising the shuffle partition count relieves per-task memory and multiplies the fetch fan-out. There is no setting that is safe in both directions, only one that fits the current data.
| Trigger | Symptom | Cause | Response |
|---|---|---|---|
| An executor is pre-empted or its node fails mid-stage. | A completed stage re-enters running; the job takes far longer than usual with no error. | Shuffle output is intermediate state on executor-local disk. Losing the executor loses the blocks the next stage needs. | Use an external shuffle service where the platform offers one, reduce reliance on pre-emptible capacity for shuffle-heavy stages, or materialise the post-shuffle result as a restart point. |
| A wide aggregation on a cluster with generous memory. | Tasks fail with no space left on the device; the cluster looks healthy on every other metric. | Spill goes to local disk, which is sized by instance type and not by your memory configuration. | Provision instance storage for the peak spill, reduce shuffled bytes by projecting and pre-aggregating, or raise partition count so each task spills less (Memory Pressure, Swap and the OOM Killer). |
| One key holds a large share of the rows. | One reduce task runs for many multiples of the median, or fails on memory while its peers finish in seconds. | The partitioner faithfully sends every row of that key to one task. The shuffle is working correctly; the distribution is the problem. | Salt the hot key, split it out and handle it separately, or eliminate the shuffle with a broadcast join (Salting a Skewed Key). |
| Tens of thousands of map tasks and tens of thousands of reduce tasks. | High fetch wait, low CPU, and a stage whose duration is dominated by requests rather than bytes. | M x R fetch requests and their bookkeeping, not the data volume. | Reduce task counts on both sides so that each does substantial work, and coalesce after selective filters instead of carrying the original partition count forward (Partitions: the Unit of Parallelism). |
| Raising the shuffle partition count to fix memory pressure. | Memory errors stop; fetch wait and stage overhead rise, and the total runtime does not improve. | The mitigation traded one shuffle cost for another. Each partition is smaller and there are many more of them. | Size from shuffled bytes per task rather than from the failure that prompted the change, and check that the fix moved the runtime rather than just the error message (The Bottleneck Moves After Every Fix). |
How to build it
Most important first.
- Shrink the rows before the shuffle. Project only the columns the downstream operation needs and apply every filter that the semantics allow first — the shuffle cost is bytes, and bytes are rows times width (Predicate Pushdown).
- Aggregate partially before redistributing. Sums, counts, minimums, maximums and approximate distinct sketches all combine, so one partial per key per partition crosses the network instead of every row (Parallel Reduce).
- Remove the shuffle entirely where one side is small: a broadcast join replicates the small side to every executor and the large side never moves (Broadcast Joins).
- Reuse an existing partitioning. If both sides of a join are already bucketed on the join key, or a second aggregation uses the same key as the first, the engine can skip the redistribution — which is what bucketing is for (Bucketing).
- Size the shuffle partition count from the volume being shuffled, not from a default, and revisit it as the data grows (Partitions: the Unit of Parallelism).
- Treat every wide operation as a deliberate cost. A defensive
DISTINCT, a window function on a different key, a sort nobody asked for — each is a full redistribution of the dataset (Narrow and Wide Transformations).
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.
- All rows with the same key land in the same output partition, for a fixed partition count and a deterministic partitioner. That is the shuffle's entire promise and everything downstream depends on it.
- It does not guarantee order within a partition unless the operation requested sorting. Rows arrive in fetch order, which varies between runs.
- It does not guarantee a stable key-to-partition mapping across runs with different partition counts. Change the count and every key may move — the same property that makes changing a topic's partition count consequential (Topics and Partitions).
- It does not guarantee that the partitions are balanced. Balance is a property of the key distribution, and the shuffle faithfully reproduces whatever imbalance the data has (Data Skew).
- Shuffle output is durable only for the lifetime of the executor that wrote it. It is intermediate state on local disk, not a checkpoint (Checkpointing).
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 expected row-count relationship across the shuffle: an aggregation must emit at most one row per distinct key, and a join must be checked against an expected fan-out. A join that emits more rows than either input has is the classic silent grain error (Grain: What Does One Row Represent?).
- Alert on shuffle bytes per run relative to input bytes. A ratio that jumps is the earliest visible sign that a plan changed — a broadcast that stopped happening, a filter that stopped pushing down (Pipeline Metrics).
- Neither check notices that the key itself is wrong. Shuffling perfectly on
customer_idwhen the grain requirescustomer_id, order_dateproduces a clean, fast, well-balanced job with a wrong answer.
- The shuffle is a barrier, so it converts the distribution of map task durations into a single number: the maximum. A job with one slow map task has a shuffle that starts late for everybody (Straggler Tasks).
- Its duration scales with bytes moved rather than rows read, which is why a job can get dramatically slower after a change that reads exactly as much data as before.
- Removing a shuffle — by broadcasting, by pre-aggregating, or by reusing a partitioning — usually does more for a job's wall clock than any amount of extra capacity (Cost vs Freshness).
- A new column widens every shuffled row, so shuffle cost grows with schema width even when nothing about the query changed (Schema Evolution).
- A key's cardinality changing — a tenant merging, a country field defaulting to a single value after a migration — changes the distribution, and the shuffle amplifies it into a straggler (Semantic Changes).
- Changing shuffle partition count changes the physical placement of every key, which silently invalidates any downstream assumption about output files (Bucketing).
- Recovery from a lost fetch is automatic and expensive: the engine re-runs the map tasks whose output is missing, which means an already-completed stage runs again (Stages and Tasks).
- There is nothing to repair in a shuffle after the fact — it holds no persistent state. What is repairable is the job's output, and that is governed by the commit protocol rather than by the shuffle (Atomic Publish).
- For jobs where recomputation is unacceptable, materialise the post-shuffle result to storage and treat it as a restart point rather than relying on lineage (Reprocessing vs Retrying).
What can go wrong
- Fetch failure: a reducing task cannot get a block, the stage retries, and the producing stage recomputes.
- Local disk exhaustion from spill, which fails tasks on a cluster that has ample memory and object storage.
- One reducing task receiving a dominant key and running out of memory, or running for many multiples of its peers (Data Skew).
- M x R fetch metadata overwhelming the cluster when both map and reduce task counts are very large — a failure of the shuffle's bookkeeping rather than of its data movement.
- The mitigation failing: raising shuffle partition count to relieve memory pressure, which makes each partition smaller and the M x R fan-out larger, and moves the bottleneck rather than removing it (The Bottleneck Moves After Every Fix).
- "The shuffle is a network transfer." It is a sort, a serialization, two disk writes, a network transfer and a merge. The network is frequently not the dominant term (Disk and Storage: Latency, Throughput, IOPS and the fsync Tax).
- "Shuffle cost is proportional to input size." It is proportional to the bytes the plan chooses to move, which a projection or a partial aggregation can reduce by orders of magnitude without changing the input at all.
- "More partitions will fix the memory pressure." It will, until the fan-out cost exceeds what it saved. Both directions have a wall.
- "Shuffle output is safe once the stage completes." It lives on executor local disks. Losing the executor loses the data and re-runs the stage (Checkpointing).
- "Avoid all shuffles." Some questions require a global regrouping and no amount of cleverness removes it. The goal is to shuffle once, over the smallest possible rows.
Operating it
- Shuffle write bytes and shuffle read bytes per stage. This pair is the primary vital sign of a distributed job (Pipeline Metrics).
- Spill to disk and spill to memory per stage, which tell you whether the reducing side fits its budget (Memory Pressure, Swap and the OOM Killer).
- Fetch wait time as a share of task time — high fetch wait with low CPU means the stage is network- or straggler-bound rather than compute-bound (Saturation: The Reading Utilization Cannot Give You).
- Maximum versus median shuffle-read bytes per task, which is the most direct measurement of skew that exists (Tail Latency: Why p50 Being Fine Does Not Help).
- At 10x data, shuffle bytes grow roughly 10x and the partition count should grow with them. Left fixed, the stage crosses from in-memory to disk-bound and the runtime grows faster than the data.
- At 100x, the shuffle is the job. Plan-level changes that remove a redistribution are worth more than any hardware decision, and jobs are often restructured to shuffle once rather than three times (Broadcast Joins).
- Executor count scales the M x R fan-out quadratically in the worst case, so very wide clusters make the shuffle's bookkeeping a first-class problem rather than an afterthought.
- Bytes serialized, written to local disk, transferred, read back and deserialized — the same rows paying four or five separate costs, which is why shuffle bytes are worth much more than an equal number of scanned bytes (What Actually Drives Data Platform Cost).
- Cluster time spent waiting at the barrier, billed across every idle executor (Compute Waste).
- Local disk provisioning: shuffle-heavy jobs need instance storage sized for the peak spill, and that requirement is invisible until a job fails on a cluster with plenty of everything else.
- Cross-zone network transfer where executors span availability zones, which is a real and frequently overlooked line (Egress: Moving Data Costs Money, Not Just Storing It).
- Pre-aggregating before the shuffle is nearly always right and costs memory on the map side plus a change in floating-point summation order, which matters when reconciliation is expected to match to the cent (Reduction Ordering: The Sum Changed When the Worker Count Did).
- Avoiding shuffles by bucketing the data costs a fixed layout that must be maintained, and benefits only the joins that use exactly that key (Bucketing).
- Larger shuffle partitions reduce fan-out and increase memory pressure per task; smaller ones do the reverse. There is no setting that is best on both axes, only one that is right for the current volume.
Shuffle lab — shuffle join against broadcast join
Change an input and watch which number moves — and which one does not. Everything here comes from a model in this repository, not from a measurement.
| Shuffle hash join | Broadcast join | |
|---|---|---|
| Moves | Both sides, once, partitioned by the join key. | The small side only — but once per task. |
| Bytes | 29.8 GBsim | 915.5 MBsim |
| Skew exposure | Total. Every row of the hottest key hashes to one reducer. 35% on one task | None from the join. The large side is never repartitioned, so it keeps whatever distribution it had. |
| Fails when | A single key is too large for one reducer's memory — the failure salting exists to postpone. | The small side stops being small. fits in 2.0 GB per task |
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.
- GENERALEvery distributed system that groups or joins by key must redistribute rows, so the map-partition-fetch-merge structure appears in Spark, Flink, Trino, MapReduce and every distributed warehouse. What differs is whether the intermediate is materialised to disk, streamed, or handled by a separate shuffle service.
- ENGINE-SPECIFICSpark materialises shuffle output to executor-local disk, which enables recomputation-based recovery and makes executor loss expensive. Trino streams between stages in memory, so it is faster and loses the whole query when a worker dies; Flink exchanges records continuously and relies on checkpoints instead.
- SIMPLIFIEDThe description here uses hash partitioning and a sort-merge reduce because they are the common case. Range partitioning for global sorts, hash-based aggregation without sorting, and push-based shuffle services all change the mechanics while leaving the cost drivers — serialization, disk, network, barrier — exactly where they are.
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 an all-to-all exchange is fundamentally different from point-to-point communication, and what a barrier costs when any participant can be slow or absent.
- — DevOps / Production Engineering owns the instance and storage configuration that decides how much spill a node can absorb before a shuffle-heavy job starts failing on a resource nobody was watching.