Spark shuffle architecture — sort spill, push-based shuffle, AQE

The wide dependency that forces a shuffle

Spark's execution graph is built out of two kinds of edges. A narrow dependency means each parent partition feeds exactly one child partition: map, filter, mapPartitions, a projection, a partition-local coalesce. Narrow edges cost nothing at the graph level -- they are fused into a single task and, in Spark SQL, into a single generated Java method, so records move from one operator to the next in CPU registers.

A wide dependency means a child partition needs records from many parent partitions. Every GROUP BY on a non-partitioning key, every sort-merge join, every repartition, distinct, and every window function whose PARTITION BY does not already match the data layout creates one. The records have to be physically redistributed so that all rows sharing a key end up in the same task. That redistribution is the shuffle.

The DAGScheduler cuts the job at exactly these edges. Everything upstream of a wide dependency becomes a ShuffleMapStage whose only product is shuffle output; everything downstream becomes another ShuffleMapStage or the final ResultStage. The cut is a full materialisation barrier: no reduce task may start until every map task in the upstream stage has finished writing, because the reducer needs to fetch from all of them. One straggling map task stalls the entire downstream stage, and no amount of downstream parallelism helps.

The barrier is also the reason Spark suddenly knows the truth about its data. At the moment a map stage completes, per-partition sizes are not estimates -- they are file lengths. That is the fact adaptive query execution is built on.

Where the reducer learns what to fetch

Each map task, on completion, reports a MapStatus to the driver's MapOutputTracker: the BlockManagerId of the executor that wrote the output, plus the size of each of the R reduce slices it produced. A reduce task asks the tracker for the locations of partition k and gets back the list of every map output that contains bytes for it.

Those sizes are stored compressed, and the compression scheme is worth knowing because it leaks into your tuning. Below spark.shuffle.minNumPartitionsToHighlyCompressSize (2000) each block size is a single byte on a logarithmic scale -- lossy but per-block. Above it, Spark switches to a highly-compressed form that keeps only a bitmap of empty blocks, one average size for the rest, and exact sizes for blocks above spark.shuffle.accurateBlockThreshold (100 MB). The reason is driver memory: with M map tasks and R reduce partitions the tracker holds M x R numbers, and at M = R = 20,000 that is 400 million of them. The consequence is that at very high partition counts, AQE's view of partition sizes is an average plus a few outliers rather than a true histogram, so skew detection gets blunter exactly where you are most likely to need it.

The map side: partition, sort, spill, index

A map task does not send anything anywhere. It writes its entire output to local disk and reports that it is done. Fetching is the reducer's problem. The write path has three moving parts.

Choosing the target partition

The partitioner turns a key into a reduce partition id. The RDD default is HashPartitioner: a non-negative modulo of key.hashCode by the partition count. Spark SQL uses a Murmur3 hash over the partitioning expressions with the same pmod semantics. The important property is that the mapping is purely a function of the key -- so a key that appears 400 million times lands 400 million records in one partition, and no writer-side cleverness can undo that. Global ordering uses a RangePartitioner instead, which reservoir-samples the input to build range bounds before the real job runs; that sampling is a separate pass over the data and shows up as a mysterious extra job in the UI.

Which shuffle writer your stage gets

Spark picks one of three writers per stage, and the choice changes the cost profile substantially.

BypassMergeSortShuffleWriter is used when there is no map-side aggregation and the partition count is at or below spark.shuffle.sort.bypassMergeThreshold (200). It skips sorting entirely: it opens one output stream per reduce partition, writes each record straight to its stream, then concatenates the files at the end. Fast and simple, but it holds R open files and R write buffers at once. At the default spark.shuffle.file.buffer of 32k, 200 partitions means 6.4 MB of buffer per running task before a single record is written -- which is why raising that buffer is not free.

UnsafeShuffleWriter, the serialised sort path, is used when the serializer can relocate serialised objects (Kryo, and Spark SQL's UnsafeRow serializer) and there is no map-side combine. Records are serialised once into memory pages; what actually gets sorted is an array of 8-byte longs packing the partition id and a pointer. The record bytes are never deserialised again -- not during the sort, not during the merge -- so the merge can even be done with raw file-stream concatenation when compression permits. This is the Tungsten machinery applied to shuffle.

SortShuffleWriter is the general fallback and the only one that supports map-side combine. It feeds an ExternalSorter, which buffers records in a partitioned append-only map (with combine) or a partitioned pair buffer (without), ordered by partition id and, when needed, by key.

Spilling and the two-file output

The sorter's buffer is not a fixed allocation. It repeatedly asks the executor's unified memory manager for more execution memory. When the request cannot be satisfied -- because other tasks on the executor are holding their share, or because storage memory has hit its immovable floor -- the sorter sorts what it holds, writes that run to a spill file under one of spark.local.dir's directories, and starts fresh.

At the end of the task, all spill runs plus whatever is still in memory are merge-sorted and written out as exactly two files: one data file containing every reduce partition's bytes concatenated in partition order, and a small .index file holding the byte offset where each partition's slice begins. This is the single most important structural fact about the modern shuffle. The old hash shuffle produced M x R physical files; at M = R = 2000 that is four million files per stage, which exhausts inodes and file descriptors long before it exhausts disk. Two files per map task means a reduce fetch is a seek to a recorded offset and a sequential read, and the file count grows with M alone.

The whole path in one picture

Read the diagram as three bands. The top band is the required path every shuffle takes: map tasks partition and write, a serving process exposes those files, reduce tasks fetch and merge. The middle band is the machinery layered on top of that path -- sort-based writing with spill, Tungsten's serialised representation, push-based merging, and runtime adaptation. The bottom band is what you actually control: whether the shuffle happens at all, what you measure afterwards, and the physical substrate of cores, memory, and local disk that all of it runs on.

The reduce side: fetching, and the usual OOM site

A reduce task's input is one slice from every map task in the upstream stage. ShuffleBlockFetcherIterator splits that list into local blocks -- read straight off this node's disk, no network involved -- and remote blocks grouped by source executor, then issues fetches under several simultaneous limits:

• spark.reducer.maxSizeInFlight (48 MB) caps total outstanding bytes. Spark deliberately splits this into roughly five concurrent requests, so it pulls from about five sources at once instead of draining one node at a time.
• spark.reducer.maxReqsInFlight caps the number of outstanding requests.
• spark.reducer.maxBlocksInFlightPerAddress caps how many blocks one reducer may request from a single source. This is the lever that stops a thousand reducers from collectively burying one node's shuffle service.
• spark.maxRemoteBlockSizeFetchToMem (200 MB) sends any block larger than the threshold to disk rather than into the heap. That default exists because a single oversized block used to be enough to kill a reducer.

Fetched blocks are decompressed and deserialised, then handed to whatever the reducer actually does: an ExternalAppendOnlyMap for aggregation, an ExternalSorter for sort-merge join or ordered output. Both can spill, so in principle memory is bounded.

In practice the reduce task is where jobs die, for a structural reason. A map task ever only holds its own sorted run; it can always spill and make progress. A reduce task holds fetch buffers, aggregation or join state, and the buffered side of a sort-merge join at the same time -- and spilling does not help when a single key's group must be materialised at once. A skewed join key with 200,000 matches on the left and 300,000 on the right produces 60 billion output rows from one task's input, and no spill threshold rescues that. When you see an executor OOM in a shuffle-heavy job, the reduce stage is the first place to look, and the distribution of one key is the first hypothesis.

Failures on this path are unusually expensive. If a fetch does not succeed after spark.shuffle.io.maxRetries (3) attempts spaced by spark.shuffle.io.retryWait (5 s), the task raises FetchFailedException. That is not a task-level retry: the DAGScheduler concludes the map output is gone, marks the map stage failed, re-runs the missing map tasks, and then re-runs the reduce stage. Losing one node midway through a large shuffle can cost you a stage you already paid for.

Why shuffle dominates the bill

Add up what moving 1 TB through a shuffle actually costs. Every byte is serialised, compressed, and written to local disk on the map side. Every byte is read back, transferred over the network unless it happens to be local, decompressed, and deserialised on the reduce side. If the sorter spills, a portion of it is written and read a second and third time. Compare that with a narrow stage, where a record is read once out of a columnar batch and stays in CPU registers through the whole operator chain.

So the shuffle pays four independent taxes -- disk write, network, disk read, and serialisation/compression CPU -- all of which scale linearly with data volume, and it pays a fifth in the form of the barrier, where stage wall-clock is set by the slowest task rather than the average one. This is why shuffle-heavy stages routinely account for most of a job's runtime while representing a small fraction of its logical work.

The practical consequence is an ordering of tuning questions. Codec selection and buffer sizes are third-order. The first-order question is whether the shuffle has to happen at all: a broadcast join removes it outright for large-small joins, bucketed or pre-partitioned sources let a join reuse an existing layout, and a runtime bloom filter shrinks the side that has to move. The second-order question is how much data crosses: pushing a filter before the exchange, projecting away unused columns, and using a partial aggregation (so the map side pre-combines and only per-key partials cross the wire) routinely cut shuffle volume by one to two orders of magnitude on low-cardinality group-bys.

Partition count, small reads, and connection fan-out

spark.sql.shuffle.partitions defaults to 200. That number is a constant unrelated to your data, your cluster, or your query, and it is wrong in both directions depending on the stage.

Too few and each reduce task receives an enormous slice. It spills repeatedly, merges through many passes, or dies. Two hundred partitions over 1.4 TB is 7 GB per task -- feasible only with heroic executor memory, and slow regardless.

Too many and you pay on three fronts at once. Task launch and bookkeeping overhead is a fixed cost per task, so tens of thousands of tasks that each process a few megabytes spend a meaningful share of their life being scheduled. The driver's map output tracker grows as M x R. And most importantly the individual fetch gets small: M map tasks and R reduce partitions produce M x R logical slices, so a 100 GB shuffle at M = R = 2000 means four million slices averaging 25 KB. A shuffle service serving 25 KB reads is doing random I/O, not streaming I/O, and its throughput collapses accordingly -- this is the exact pathology push-based shuffle was built to fix. If the stage is also the final write, R small partitions become R small output files, and the downstream job inherits your partition count as its file count.

A workable target is 100-200 MB of shuffle read per reduce task, derived from the previous run's total shuffle read rather than guessed. Round it to a multiple of total executor cores so the last wave of tasks is not half empty. And rather than hand-tuning per stage, set the value high as a ceiling and let AQE coalesce downward, which is what it is for.

Reading the shuffle metrics

Every diagnosis starts with the stage detail page. Here is a real-shaped example of a job that is not doing well:

Stage 7  (ShuffleMapStage)  2000 tasks
  Shuffle Write Size / Records     1.4 TB / 9.8 B
  Shuffle Write Time (total)       41 min
  Spill (Memory)                   3.1 TB
  Spill (Disk)                     612 GB

Stage 8  (ResultStage)       200 tasks
  Shuffle Read Size / Records      1.4 TB / 9.8 B
    local / remote                 70 GB / 1.33 TB
  Fetch Wait Time (total)          2.1 h
  Spill (Memory)   max / median    48 GB / 2 GB
  Duration         max / median    57 min / 4 min

Spill (Memory) versus Spill (Disk). These are the same data measured twice. Spill (Memory) is the deserialised, in-heap footprint of what was evicted; Spill (Disk) is the serialised and compressed bytes actually written. The ratio -- about 5x here -- is your in-memory expansion factor, and it is the number to use when you size executors against a known input size. A small disk spill is graceful degradation and not a problem. Spill several times the size of the shuffle itself means the sorter is thrashing through repeated merge passes and needs either more execution memory or more partitions.

Fetch Wait Time is time reduce tasks spent blocked with nothing to process. High fetch wait alongside low network utilisation almost never means a slow network; it means too few bytes in flight, or every reducer converging on one overloaded source node.

max versus median. The 14x duration ratio on stage 8 is the classic skew signature. Cross-check it against max/median of Shuffle Read Size: if the bytes are even but the durations are not, the culprit is one sick node or a GC pause, not the data distribution. Here the bytes will be uneven, and the underlying bug is visible in the first line -- 200 tasks reading 1.4 TB.

Tuning levers and what each one costs

Two of these deserve emphasis because they are commonly mis-set. Codec choice is a cluster-shape decision, not a preference: measure whether your shuffle stages are pinned on CPU or on disk and network before changing it, because zstd on a CPU-saturated cluster makes things worse. And the retry settings are the cheapest available insurance on a shared cluster -- a fetch failure costs you a re-run of a whole map stage, so waiting an extra thirty seconds for a stalled peer is a trade you almost always want.

Also give spark.local.dir real attention. It should point at several directories on separate fast local devices; shuffle write, spill, and fetch all land there, and putting it on a single network-attached volume converts every one of the four taxes above into a much larger one.

What AQE takes off your hands

Because the shuffle barrier produces exact per-partition sizes, the optimizer can revisit its decisions between stages. Adaptive query execution uses that to coalesce many small post-shuffle partitions into fewer right-sized tasks, to split an oversized partition into sub-partitions with the join's other side duplicated across them, and to downgrade a planned sort-merge join to a broadcast join once a side turns out to be small. In practice this converts spark.sql.shuffle.partitions from a per-query battle into a ceiling you set once.

Two limits are worth carrying with you here. AQE can only adapt what comes after a shuffle, so a bad decision before the first exchange still executes in full. And its skew handling depends on the size statistics described above, which lose resolution once the partition count crosses the highly-compressed threshold. The mechanics, the thresholds, and the plan-reading workflow are covered in the AQE article.

Push-based shuffle and the problem it solves

The default shuffle is pull-based, and its weakness is the shape of the reads rather than their volume. Each reducer pulls one small slice from each of M map outputs scattered across the cluster; at large M and R those slices shrink until the shuffle service is servicing millions of tiny random reads. Disk seek behaviour, not bandwidth, becomes the limit, and the tail of the fetch distribution stretches out.

Push-based shuffle inverts the direction. After a map task writes its normal output, it also pushes its blocks to shuffle merger services chosen by reduce partition id. Each merger appends the blocks it receives for a given partition into one merged file, so by the time reducers start, a large fraction of each reduce partition already exists as a handful of big sequential chunks in one place instead of thousands of fragments in a thousand places. Two things follow: reads become sequential and large, and the reducer gains a locality preference -- it can be scheduled on the node holding its merged partition.

The design is deliberately best-effort. Pushing is asynchronous and merging can fail, be too slow, or be skipped; when that happens the original map output is still sitting on disk exactly as before, and the reducer simply falls back to fetching the unmerged blocks. Correctness never depends on the merge succeeding, which is what makes it safe to enable. The cost is extra network during the map stage, extra disk on the merger nodes, and a dependency on having shuffle merger services available -- which in practice means it is a feature of the external shuffle service deployment rather than something you turn on in isolation. It pays for itself on large clusters running big shuffles, and does essentially nothing for small jobs.

Decoupling shuffle data from executor lifetime

Shuffle files are written by an executor but must outlive it. If the only process that can serve them is the JVM that produced them, then Spark cannot release an idle executor without destroying data a later stage needs, and every executor loss becomes a stage re-run. The external shuffle service fixes this by moving the serving role into a long-lived per-node daemon: files survive executor exit, dynamic allocation can actually scale down, and a crashed executor stops being a data-loss event. The daemon's internals, its zero-copy serving path, and its capacity limits are covered in the external shuffle service article.

On Kubernetes the classic node-level shuffle service does not exist, because there is no YARN auxiliary-service slot to host it, and the assumption it encoded -- that a node outlives the executors on it -- is exactly what an autoscaled or spot-backed cluster breaks. Three mitigations are in common use, and they are not equivalent.

Graceful decommissioning (spark.decommission.enabled with spark.storage.decommission.shuffleBlocks.enabled) has a doomed executor migrate its shuffle blocks to surviving peers before the pod goes away. It handles planned scale-down and spot preemption notices well; it does nothing for an abrupt kill, and migration itself consumes network at the worst possible moment.

PVC reuse puts shuffle data on a persistent volume claim that Spark can re-attach to a replacement executor pod, so the data survives the pod rather than the process. It preserves the work but not the availability -- nothing serves those files until a new pod mounts them.

Disaggregated remote shuffle services take the position that node-local shuffle is the wrong abstraction for elastic compute at all. Mappers push shuffle output to a separate storage tier, reducers read from it, and compute nodes hold no durable state. That is what makes aggressive autoscaling, heavy spot usage, and instant scale-to-zero safe; the price is a service to run and a network round-trip that was previously a local disk read.

Whichever route you take, size and site the scratch storage deliberately. A pod's emptyDir backed by the node's local NVMe behaves roughly like a traditional local disk; the same emptyDir with a memory medium is tmpfs, which quietly consumes the pod's memory limit and turns a spill into an OOM kill; and a network-attached block volume makes every shuffle write, spill, and fetch dramatically slower than the numbers your on-premise intuition expects.