Spark shuffle internals: the three writers, and which one your job gets

• Why does a shuffle exist?
• Which operations shuffle, and what does it cost?
  • Where the cost actually goes
• What should you know already?
• How is the shuffle subsystem put together?
• What decides which writer you get?
  • The serializer is picked by your record types
• What lands on disk?
• How do the three writers differ?
• What does spilling actually cost?
• What does the reduce side actually do?
• What happens when a fetch fails?
• Who serves the files when the executor is gone?
• Which numbers actually tell you something?
• What does adaptive execution change about a shuffle?
• How do you avoid the shuffle entirely?
• What should you actually configure?
• Symptoms, causes and fixes
• Why did hash shuffle lose, and what does push-based change?
• When is the shuffle not your problem?
• Common misconceptions
• Interview questions
  • What a shuffle is, and what it costs
  • Which writer you get
  • What lands on disk
  • The reduce side
  • Spill, memory and tuning
• A model worth keeping
• References
• Trademarks

TL;DR

• Sort-based shuffle writes one data file per map task, not one per map-and-reduce pair, alongside an index of partition offsets and a checksum file. Collapsing that file count is what let the shuffle scale.
• Three writers exist: BypassMergeSortShuffleWriter, UnsafeShuffleWriter and SortShuffleWriter. SortShuffleManager.registerShuffle picks one per shuffle, and nothing in the UI tells you which.
• The bypass path requires no map-side aggregation and at most spark.shuffle.sort.bypassMergeThreshold partitions, default 200. The comparison is <=: 200 partitions bypass, 201 do not.
• The serializer is chosen by your record types, not only by spark.serializer. Kryo is selected automatically when both key and value are primitives, primitive arrays or String; anything else falls back to the configured default.
• Thirty interview questions at the end, including a one-liner that prints which of the three writers a shuffle actually got — the Spark UI never tells you.
• That makes the writer type-dependent. Changing a value from Int to (Int, Int) moves the same job off UnsafeShuffleWriter and onto SortShuffleWriter, which pays to deserialize and re-serialize records when it merges spills.

Why does a shuffle exist?

A groupByKey has to put every record sharing a key in one place, and those records are spread across every executor that read the input. Nothing in the upstream stage arranged them that way, and no amount of local computation can fix it: the data has to move.

That movement is the shuffle, and it is the only mechanism Spark has for relocating records between partitions. Everything else in the execution model is local. A map, a filter or a mapPartitions transforms a partition in place, so the task reads one parent partition and writes one child partition. Spark calls that a narrow dependency and fuses the whole chain into a single stage.

A wide dependency breaks the chain, because a child partition draws from every parent partition. Spark cannot fuse across that boundary, so it cuts the plan there and the cut becomes a stage boundary. The stages either side are not two halves of one computation running together; the first one runs to completion and writes its output to local disk, and only then does the second start reading.

This is worth stating plainly, because it explains most of what follows. The shuffle is a disk barrier, not a network transfer. Map tasks write files. Reduce tasks fetch from those files afterwards. Nothing streams directly from one task to another. That design buys the thing Spark needs most at scale: a reduce task that dies can be retried by re-fetching the same bytes, without re-running the stage that produced them.

It also sets up the cost. Every shuffle writes the full intermediate result to disk, serializes it on the way out, and deserializes it on the way back in. When a job is slow, the shuffle is usually where the time went.

The earlier design made that cost worse in a way that did not survive contact with large clusters. Hash-based shuffle gave each map task one open file per reduce partition, so a stage of M map tasks feeding R reduce partitions produced M × R files. At a thousand map tasks and a thousand reduce partitions, that is a million files and a million write buffers, and the job died on file descriptors or drowned in random I/O long before it ran out of compute. Sort-based shuffle replaced it by having each map task write a single file with its records sorted by partition id, plus a small index recording where each partition starts. M files, not M × R.

That is the design every current Spark job uses, and it comes in three implementations. They differ in whether records are sorted as objects or as serialized bytes, whether a merge step happens at all, and what they cost in memory and CPU. Which one a given shuffle gets is decided once, when the shuffle is registered, from properties of the dependency rather than anything you set directly, and the choice is never reported. The rest of this post is that decision and its consequences.

Which operations shuffle, and what does it cost?

Not every wide operation announces itself. These all introduce a shuffle:

coalesce is the exception worth remembering: it merges existing partitions without redistributing rows, so it narrows without a shuffle.

The cut itself is made by the DAGScheduler, which walks the RDD lineage and breaks it at every wide dependency. The stage that writes shuffle output is a ShuffleMapStage, whose tasks are ShuffleMapTasks; the final stage that produces the job’s result is a ResultStage. A job is therefore some number of ShuffleMapStages feeding one ResultStage, and the count of shuffles in a job is exactly the count of ShuffleMapStages.

Where the cost actually goes

“Shuffles are expensive” is true and useless. The expense has four distinct components, and different symptoms point at different ones.

Disk I/O. Every map task writes its entire output to local disk, and every reduce task reads it back. This is not a cache or an overflow; it is the normal path. A stage shuffling 100 GB writes 100 GB and reads 100 GB, before any spill.

Network I/O. Reduce tasks fetch from every map task, so a shuffle is an all-to-all exchange. With M map tasks and R reduce tasks there are up to M × R block transfers, which is why block count matters independently of byte volume, and why push-based shuffle exists.

Memory and GC. Records are serialized on the way out and deserialized on the way in, and sorters hold buffers that grow until they spill. The 10× gap between memoryBytesSpilled and diskBytesSpilled measured later in this post is exactly this cost made visible.

CPU. Serialization, compression (spark.shuffle.compress is on by default) and sorting all consume CPU that is doing no analytical work. This is the component people forget, and it is why a shuffle can be slow on a cluster whose disks and network are both idle.

What should you know already?

This post assumes you have written Spark jobs and read a stage graph, but not that you have opened the shuffle source. You should be comfortable with RDDs and DataFrames, partitions, the driver and executor split, and the idea that a job is cut into stages.

Three terms are used throughout in their precise sense. A map task is a task in the stage that writes shuffle output; a reduce task is one in the stage that reads it. Spark names them by position, not by operator, so a single stage is a reduce stage for the shuffle below it and a map stage for the shuffle above it. A partition id is the number the partitioner assigns to a record, which decides which reduce task eventually reads it. And map-side aggregation, or map-side combine, is the optimisation where reduceByKey folds records for the same key together before they are written, so less data crosses the boundary.

Everything below was read at the v4.1.3 tag and run on Spark 4.1.3 with Scala 2.13.17 on JDK 21, in local mode. Where the mode matters, it is called out.

How is the shuffle subsystem put together?

Five components cooperate, and each one has a single job. Knowing which is which turns most shuffle configuration from arbitrary into obvious.

flowchart TB
    subgraph DRIVER["Driver"]
        DAG[DAGScheduler] -->|registers| SM[SortShuffleManager]
        SM --> H{ShuffleHandle}
        MOT[MapOutputTrackerMaster]
    end
    subgraph MAP["Executor, map task"]
        W[ShuffleWriter] --> RES[IndexShuffleBlockResolver]
        RES --> D[(data + index<br/>+ checksum)]
    end
    subgraph RED["Executor, reduce task"]
        R[BlockStoreShuffleReader] --> FI[ShuffleBlockFetcherIterator]
    end
    H -->|"chooses the writer"| W
    W -->|"MapStatus: sizes per partition"| MOT
    MOT -->|"where each block lives"| FI
    D -->|"fetched by offset"| FI

SortShuffleManager is the entry point, and it exists once per application. It is asked to register a shuffle when the dependency is created, and asked for a writer when each map task starts. Registration is where the decision this post is about gets made.

The ShuffleHandle is how the decision travels. It is a small object created at registration time and carried to every map task, and its type is the decision: BypassMergeSortShuffleHandle, SerializedShuffleHandle or BaseShuffleHandle.

The ShuffleWriter does the work in the map task. There are three implementations, one per handle type, and they are the subject of most of this post.

IndexShuffleBlockResolver owns the files. It decides their names, writes the index of partition offsets, and performs the atomic rename that makes a map task’s output visible. Every writer goes through it, which is why the on-disk layout is identical whichever writer ran.

MapOutputTracker is the directory. Map tasks report a MapStatus holding the size of each partition they wrote; the reduce side asks the tracker where the blocks for its partition live, and fetches them. The tracker living on the driver is what makes a shuffle recoverable: the files can be found again after a task dies, and it is also why losing an executor is expensive, since its map output goes with it.

What decides which writer you get?

SortShuffleManager.registerShuffle runs one if/else if/else, once per shuffle, and the branch taken is fixed for the life of that shuffle.

flowchart TB
    S[registerShuffle] --> B{"map-side combine?<br/>partitions &le; bypassMergeThreshold?"}
    B -->|"no combine, few partitions"| BH["BypassMergeSortShuffleHandle<br/>→ BypassMergeSortShuffleWriter"]
    B -->|otherwise| C{"serializer relocatable?<br/>no combine?<br/>partitions &le; 16777216?"}
    C -->|yes| SH["SerializedShuffleHandle<br/>→ UnsafeShuffleWriter"]
    C -->|no| BS["BaseShuffleHandle<br/>→ SortShuffleWriter"]

The two predicates are small enough to read in full. shouldBypassMergeSort refuses when there is map-side aggregation, and otherwise compares the partition count against spark.shuffle.sort.bypassMergeThreshold, which defaults to 200:

def shouldBypassMergeSort(conf: SparkConf, dep: ShuffleDependency[_, _, _]): Boolean = {
  // We cannot bypass sorting if we need to do map-side aggregation.
  if (dep.mapSideCombine) {
    false
  } else {
    val bypassMergeThreshold: Int = conf.get(config.SHUFFLE_SORT_BYPASS_MERGE_THRESHOLD)
    dep.partitioner.numPartitions <= bypassMergeThreshold
  }
}

The comparison is <=, and that boundary is exact rather than approximate. A shuffle with 200 partitions takes the bypass path; the same shuffle with 201 does not.

canUseSerializedShuffle asks three questions, and the first is the one that surprises people:

if (!dependency.serializer.supportsRelocationOfSerializedObjects) {
  false          // the serializer cannot be trusted to move serialized bytes
} else if (dependency.mapSideCombine) {
  false          // combining needs the objects, not their bytes
} else if (numPartitions > MAX_SHUFFLE_OUTPUT_PARTITIONS_FOR_SERIALIZED_MODE) {
  false          // 16777216, from the 24 bits the packed pointer gives the partition id
} else {
  true
}

Relocation is the crux. The serialized path sorts records without deserializing them, which means moving an already-serialized record from one position in a buffer to another and expecting it to still be readable. A serializer that writes stream-level state, such as a back-reference table, breaks under that treatment. JavaSerializer writes exactly that kind of state and reports false; KryoSerializer reports true.

The partition ceiling of 16777216 comes from PackedRecordPointer, which encodes a partition id and a memory address in a single 64-bit long, giving the partition id 24 bits.

The serializer is picked by your record types

The part that is easy to get wrong: spark.serializer is not the whole answer. ShuffledRDD asks SerializerManager for a serializer, and it auto-selects Kryo when both the key and value class tags are primitives, primitive arrays or String:

def getSerializer(keyClassTag: ClassTag[_], valueClassTag: ClassTag[_]): Serializer = {
  if (canUseKryo(keyClassTag) && canUseKryo(valueClassTag)) {
    kryoSerializer
  } else {
    defaultSerializer
  }
}

spark.serializer sets defaultSerializer, which still defaults to org.apache.spark.serializer.JavaSerializer. So the configured default only applies to the records the auto-pick rejects.

Running the real decision functions over six record types, all at 300 partitions so only the type varies, gives:

A tuple is not a primitive, so (Int, (Int, Int)) falls off the fast path although every leaf in it is an Int. Wrapping two counters in a case class does the same. Nothing in the plan, the UI or the logs reports that the writer changed.

This matters more for RDD code than for SQL. Spark SQL shuffles UnsafeRow through UnsafeRowSerializer, which supports relocation by construction, so DataFrame and Dataset jobs reach the serialized path regardless of the column types. The type sensitivity above is an RDD-level concern.

What lands on disk?

Whichever writer ran, the output is the same shape: one data file per map task, an index of offsets, and a checksum file.

<spark.local.dir>/blockmgr-<uuid>/
└── <hash prefix>/
    ├── shuffle_0_0_0.data              # every partition's bytes, concatenated
    ├── shuffle_0_0_0.index             # numPartitions + 1 offsets, as longs
    ├── shuffle_0_0_0.checksum.ADLER32  # one checksum per partition
    ├── shuffle_0_1_0.data
    ├── shuffle_0_1_0.index
    └── ...

The name is shuffle_{shuffleId}_{mapId}_{reduceId}, and the trailing component is always 0. It is IndexShuffleBlockResolver.NOOP_REDUCE_ID, a placeholder kept so the block id format still has three parts. Its constancy is the whole point of sort-based shuffle: there is no per-reducer file.

A shuffle of 8 map tasks into 4 partitions produced 24 files, and the sizes are worth checking rather than trusting:

The index holds one offset per partition plus a final end-of-file offset, as 8-byte longs, which is why a reduce task can seek straight to its own bytes with two reads of the index. The checksum file exists because spark.shuffle.checksum.enabled defaults to true with spark.shuffle.checksum.algorithm set to ADLER32; it lets Spark tell a genuine network corruption from a bug when a fetched block will not decompress.

A map task writes its data and index to temporary files and then calls IndexShuffleBlockResolver.writeMetadataFileAndCommit, which renames them into place under a lock. That rename is the commit. It is also what makes speculative execution safe: two attempts at the same map task can both write, and whichever commits first wins, with the loser’s temporary files discarded.

How do the three writers differ?

All three produce that identical layout. What differs is the path taken to it.

BypassMergeSortShuffleWriter does not sort at all. It opens one DiskBlockObjectWriter per reduce partition, writes each record straight to the file for its partition, then concatenates those files in partition order into the single data file and records the lengths as the index. The name is precise: it bypasses the merge sort by never producing anything that needs merging.

The cost is in the first sentence. It holds numPartitions files open at once, each with its own buffer of spark.shuffle.file.buffer, which defaults to 32k. At 200 partitions that is 200 open handles and roughly 6 MiB of buffer per task, which is exactly why the threshold that guards it defaults to 200 rather than something larger.

UnsafeShuffleWriter sorts serialized bytes. Records are serialized on arrival and appended to memory pages; what gets sorted is an array of PackedRecordPointer longs, each packing the partition id into 24 bits and the record’s address into the rest. Sorting moves those 8-byte longs, never the records.

That buys two things. Sorting touches a compact array rather than a graph of objects, so it is cache-friendly and produces no garbage per record. And because the spilled files are already serialized and sorted by partition, merging them can be done on bytes: mergeSpills picks mergeSpillsWithTransferTo when spark.file.transferTo allows it and encryption is off, so the merge runs as transferTo calls between file channels and the records never re-enter the JVM heap. When that is not available it falls back to mergeSpillsWithFileStream.

SortShuffleWriter sorts objects. It is the general path, used whenever the other two are refused, and it is the only one that can do map-side aggregation, because combining values for a key requires the values as objects. It inserts records into an ExternalSorter, which spills sorted runs to disk under memory pressure and merges them at the end.

Which structure the sorter fills depends on whether it is combining:

aggregating (reduceByKey)     -> PartitionedAppendOnlyMap
not aggregating (partitionBy) -> PartitionedPairBuffer

The map can update a key’s value in place, which is what map-side combine needs. The buffer only appends, which is cheaper when there is nothing to combine. Both are partition-aware, so the sort key is (partition id, key) rather than the key alone.

The merge is where its cost lives. Every spilled run has to be deserialized to be merged and re-serialized on the way out, so the records make two extra trips through the serializer that neither other writer pays.

What does spilling actually cost?

A sorter spills when it cannot get more execution memory. The records written so far are sorted, written to a temporary file, and the in-memory structure is emptied; at the end, the runs are merged. Spilling is not a failure, and a job that spills a little is usually fine. What it costs is worth measuring rather than guessing.

Running a reduceByKey over 12,000,000 records with 6,000,000 distinct keys, in four map tasks with spark.memory.fraction cut to 0.1 to force the issue, the map stage reported:

The two spill metrics differ by a factor of ten, and they are not measuring the same thing. memoryBytesSpilled is the estimated size of the records as live JVM objects; diskBytesSpilled is what those records occupied once serialized and compressed, since spark.shuffle.spill.compress defaults to true. The gap between them is the cost of representing a (Int, Long) pair as boxed objects with headers and references rather than as sixteen bytes.

Two practical consequences follow. A stage reporting hundreds of megabytes of memory spill has not necessarily written hundreds of megabytes to disk, so reading memoryBytesSpilled as disk traffic overstates the I/O. And the ratio is a direct measure of how much your record representation is costing you, which is the strongest argument for pushing RDD work with rich object types into DataFrames, where the same data lives in UnsafeRow and never inflates like this.

Note also that the shuffle wrote 12,000,000 records although reduceByKey performs map-side aggregation over 6,000,000 distinct keys. No combining happened, because each input partition held a contiguous range of i and every key within a partition was therefore distinct. Map-side combine reduces nothing when keys do not repeat within a partition, which is a useful reminder that its benefit depends on how the data was partitioned upstream, not on the global cardinality.

What does the reduce side actually do?

The write side gets most of the attention; the read side is where a reduce task spends its memory.

Locations come from the driver. MapOutputTracker is asked which blocks hold this partition and where they live. The answer includes sizes, which is what lets the reader skip empty blocks entirely rather than opening them.

ShuffleBlockFetcherIterator does the fetching, and it is deliberately not a simple loop. It separates local blocks, read straight from disk with no network involved, from remote ones, and it issues remote requests concurrently while bounding how much data is in flight at once. That bound is spark.reducer.maxSizeInFlight, default 48 MiB, and it exists because a reduce task fetching from a thousand executors would otherwise try to hold a thousand responses at once.

Then the records are combined or sorted, by two different mechanisms. This distinction is worth getting right, because both classes get called “the external map” loosely:

ExternalAppendOnlyMap is the read-side twin of the write-side structures: an in-memory map that spills to disk when execution memory runs out and merges the spills on read. If map-side combine already ran, the reader merges partial results with combineCombinersByKey; if it did not, it combines raw values with combineValuesByKey.

The practical consequence: a reduce task can spill even when the map side did not. Aggregating a billion distinct keys builds a large map in the reduce task regardless of how tidily the map side wrote its output.

What happens when a fetch fails?

Reduce tasks fetch blocks over the network, and the failure handling is layered.

Transient network errors are retried inside the fetcher. The transport layer retries a failed fetch up to spark.shuffle.io.maxRetries times, which defaults to 3, waiting spark.shuffle.io.retryWait between attempts. The task never sees these; they are invisible in the UI except as time.

Corruption is detected and treated separately. With spark.shuffle.detectCorrupt on by default, a block that fails to decompress is retried, and the checksum file is what lets Spark distinguish a block corrupted in flight from one that was written wrong.

A missing block becomes a FetchFailedException, and this is the one that changes the shape of the job. The reduce task fails, the DAGScheduler marks the map output for that executor as lost, and it re-runs the map stage to regenerate the missing files before retrying the reduce stage. This is why losing one executor late in a long job can cost far more than the work that executor was doing: its map output went with it, and every shuffle that depended on it has to be partially recomputed.

The string to grep for is FetchFailedException, and the line in the driver log that precedes the re-run names the executor and the shuffle id.

Who serves the files when the executor is gone?

Shuffle files live on the local disk of the executor that wrote them, and they are served by that executor’s own process. That couples two things which have no business being coupled: the lifetime of a compute process and the availability of data other tasks still need.

It has two consequences. An executor that dies takes its map output with it, so the FetchFailedException path above re-runs the map stage. And an idle executor cannot be reclaimed, because something may still need to fetch from it.

The External Shuffle Service breaks the coupling. It is a separate long-lived process on each node, serving shuffle files from local disk independently of the executors that produced them. Turn it on with spark.shuffle.service.enabled, which defaults to false, and it listens on spark.shuffle.service.port, default 7337. It is a cluster-level deployment decision, not something a single job enables: the service has to be running on the nodes.

With it, an executor can exit while its shuffle output stays fetchable, which is what makes dynamic allocation safe. spark.dynamicAllocation.enabled defaults to false; when on, Spark adds executors under load and removes idle ones.

The part that is commonly out of date: dynamic allocation no longer strictly requires the shuffle service. spark.dynamicAllocation.shuffleTracking.enabled defaults to true, and it makes the driver track which executors hold shuffle data still referenced by an active job, keeping exactly those alive and releasing the rest. That is what allows dynamic allocation on Kubernetes, where running a node-level shuffle service is awkward.

The trade between them is real. Shuffle tracking needs no extra infrastructure but keeps executors alive while their output is referenced, so reclamation is slower. The shuffle service reclaims aggressively but is another process to deploy, monitor and upgrade in lockstep.

Which numbers actually tell you something?

The stage detail page and the TaskMetrics behind it carry more than most people read. These are the ones that answer a question:

Compare records written to records read before anything else. They should agree for the whole shuffle, and the per-task distribution of records read is where skew shows up: a stage where the maximum task reads twenty times the median is skewed, and no amount of memory tuning will fix it. That belongs to data skew rather than here.

A large memoryBytesSpilled with a small diskBytesSpilled is the signature of expensive record types, as the arithmetic above shows.

What does adaptive execution change about a shuffle?

Everything so far describes a shuffle that is going to happen. Adaptive query execution re-plans after the map side has written, using sizes it has measured rather than estimated, and three of its behaviours act directly on shuffles. All three are on by default.

Coalescing partitions. spark.sql.adaptive.coalescePartitions.enabled merges small post-shuffle partitions into fewer, larger reduce tasks, targeting spark.sql.adaptive.advisoryPartitionSizeInBytes, default 64 MiB, and never going below coalescePartitions.minPartitionSize, default 1 MiB. This is why spark.sql.shuffle.partitions matters less than it used to: the configured number is what gets written, not what gets read. It shows in a plan as AQEShuffleRead coalesced.

Splitting skewed partitions. spark.sql.adaptive.skewJoin.enabled detects a partition that is both larger than skewedPartitionThresholdInBytes, default 256 MiB, and more than skewedPartitionFactor, default 5.0, times the median partition size. Both conditions must hold, which is why a skewed job with partitions under 256 MiB gets no help from it. The fix splits the oversized partition into sub-partitions and replicates the matching side.

Switching the join strategy. Once the real post-shuffle sizes are known, a sort-merge join whose build side turned out to be small can be replanned as a broadcast hash join, and localShuffleReader.enabled then lets it read shuffle blocks locally rather than re-fetching them across the network.

The wider behaviour of AQE, including what it declines to touch, is in adaptive query execution.

How do you avoid the shuffle entirely?

The fastest shuffle is the one that does not run. Three ways out, in order of how often they apply.

Broadcast the small side. A join whose smaller side fits in spark.sql.autoBroadcastJoinThreshold, default 10 MiB, is planned as a broadcast hash join with no shuffle on either side. This converts a wide dependency into a narrow one, and it is the single highest-value change available on a join.

Combine before the exchange. reduceByKey and aggregateByKey aggregate on the map side, so the shuffle carries one partial result per key per partition instead of every value. groupByKey carries everything. In DataFrame code the planner inserts the partial aggregate for you, which is one more reason to prefer it over RDD code.

Bucket the tables. Writing both sides bucketed on the join column by the same bucket count removes the exchange permanently, because matching rows are already co-located on disk. It is a decision about the table rather than the query, and changing the bucket count means rewriting the data.

The measured plan evidence for all three is in Spark performance tuning.

What should you actually configure?

Most shuffle tuning is one number, and it is not a shuffle setting.

Partition count dominates everything. It decides which writer you get, how many files are produced, how big each partition is, and whether reduce tasks spill. For SQL, that is spark.sql.shuffle.partitions and, more usefully, whatever adaptive query execution coalesces it to at runtime. Aim for reduce partitions in the range of a hundred megabytes or so, and prefer letting AQE settle it over pinning a number.

spark.shuffle.sort.bypassMergeThreshold is worth understanding before changing. Raising it pushes more shuffles onto the bypass path, which avoids sorting entirely but opens one file and one 32k buffer per partition per task. Raising it to 2000 on a job with 2000 partitions asks each map task for 2000 open files and around 62 MiB of buffers. The default of 200 is the point where that trade stops paying.

spark.shuffle.file.buffer at 32k is the write buffer per output file. Raising it reduces syscalls on large shuffles, and multiplies by the open file count on the bypass path, which is the same trap as above seen from the other side.

spark.reducer.maxSizeInFlight at 48m bounds how much fetched data a reduce task holds at once. Raising it can help a network-bound fetch and costs memory in every reduce task simultaneously.

spark.memory.fraction at 0.6 and spark.memory.storageFraction at 0.5 set the pool a sorter allocates from and how much of it caching may hold onto. Spilling is a symptom of this pool being too small for the partition, so raising the partition count and lowering what you cache are usually better first moves than raising the fraction. The arithmetic is in Spark memory management.

Leave the compression and checksum settings alone unless you are measuring them. spark.shuffle.compress and spark.shuffle.spill.compress both default to true and both earn it on any realistic record. The codec itself is spark.io.compression.codec, default lz4, shared with every other block Spark compresses. zstd trades CPU for a smaller footprint, which pays when the shuffle is network-bound and costs when it is CPU-bound.

spark.shuffle.io.maxRetries at 3 and spark.shuffle.io.retryWait govern how hard a fetch tries before it becomes a FetchFailedException and a stage re-run. On a busy or flaky cluster, raising these converts expensive stage recomputation into cheap waiting.

A summary of every knob named in this post, with what it actually controls:

Symptoms, causes and fixes

Stage is slow and diskBytesSpilled is large. Reduce partitions are too big for the memory each task gets. Raise the partition count so each task holds less, before touching memory fractions.

FetchFailedException, then the previous stage runs again. An executor holding map output was lost, or the shuffle service could not serve it. Look for the executor’s own failure first; the fetch failure is a symptom, and an OOM or a container kill upstream is usually the cause.

Too many small files downstream, or a slow final write. The reduce partition count is too high for the volume. This is AQE’s coalescing problem and the reason to leave it enabled.

One task in the stage takes as long as the rest combined. Skew, not shuffle configuration. Check the per-task records read distribution.

Memory spill is enormous but the shuffle write is small. Record representation, as above. Move the work to DataFrames, or flatten the types so the auto-picked Kryo path applies.

Why did hash shuffle lose, and what does push-based change?

Hash-based shuffle is gone, not deprecated: there is no HashShuffleManager in the tree, and spark.shuffle.manager accepts sort as its default. The reason is the file count. One file per map task per reduce partition means M × R files, and M × R open buffers during the write, which stops working at exactly the scale where Spark is interesting.

There was an intermediate design worth knowing about, because it shows what the real constraint was. Consolidated hash shuffle had the tasks running concurrently on one executor share a set of files, called a file group: instead of every task creating its own R files, each core’s worth of concurrent tasks appended into the same R files. That cut the file count from M × R to roughly cores × R, which helped, and it did not solve the problem, because the count still scaled with the number of reduce partitions.

Sort-based shuffle changed the exponent rather than the constant: one file per map task, with partitions as byte ranges inside it and an index to find them. A file count of M, whatever R is. That trade has never stopped being worth it.

Push-based shuffle is the current direction, not a replacement of the writer. With spark.shuffle.push.enabled, map tasks additionally push their blocks to remote shuffle services, which merge the fragments belonging to each reduce partition into larger, sequential files. The reduce side then reads a few large blocks instead of thousands of small ones. It needs a server-side MergedShuffleFileManager implementation configured through spark.shuffle.push.server.mergedShuffleFileManagerImpl, so it is a deployment-level feature rather than something to switch on in a local job.

The pluggable layer underneath is worth knowing about: spark.shuffle.sort.io.plugin.class defaults to LocalDiskShuffleDataIO, and that interface is how remote shuffle services replace local disk entirely.

When is the shuffle not your problem?

Not every slow stage with a shuffle in it is a shuffle problem.

If one task is slow and the rest are fast, that is skew, and the fix is in how keys distribute. If the stage reads far more data than it needs, the fix is pushdown and pruning at the scan, before a byte reaches the shuffle. If a join shuffles at all when one side is small, the real fix is a broadcast, which removes the shuffle rather than tuning it, and join strategy selection decides that.

And if your shuffle is small and the job is slow, the shuffle is not where the time went. Read the stage timeline before tuning anything.

Common misconceptions

“Spark uses Java serialization by default, so RDD shuffles cannot use the fast writer.” The configured default is JavaSerializer, but SerializerManager auto-picks Kryo when both key and value are primitives, primitive arrays or String, and those shuffles do reach UnsafeShuffleWriter.

“Setting spark.serializer to Kryo guarantees the serialized path.” It sets the fallback for types the auto-pick rejects. It does not help if the shuffle has map-side aggregation, and it does not apply where the auto-pick already chose.

“Sort-based shuffle sorts your data.” It sorts records by partition id, so the data file’s partitions are contiguous. Records inside a partition are not ordered by key unless a key ordering was requested.

“memoryBytesSpilled tells you how much was written to disk.” It is the in-memory object size of what spilled. The measured run above wrote about a tenth of it.

“More partitions always reduces spill.” It reduces the size of each reduce partition, which helps, but it also increases the number of blocks in the shuffle, and past the bypass threshold it changes which writer runs.

Interview questions

Thirty questions, in the order the post works through the subsystem: what a shuffle is, which writer you get, what lands on disk, what the reduce side does, and what to tune. Answers link into the measured sections rather than repeating them.

What a shuffle is, and what it costs

Which writer you get

val dep = rdd.dependencies.head.asInstanceOf[ShuffleDependency[_, _, _]]
println(dep.shuffleHandle.getClass.getSimpleName)

What lands on disk

shuffle_0_0_0.data              # the records
shuffle_0_0_0.index             # numPartitions + 1 offsets, as 8-byte longs
shuffle_0_0_0.checksum.ADLER32  # one checksum per partition

The reduce side

Spill, memory and tuning

A model worth keeping

A shuffle is a disk barrier between two stages. Map tasks each write one data file with the records sorted by partition id and an index of where each partition starts; reduce tasks look up locations from the map output tracker and fetch the byte ranges belonging to them.

Which writer produced that file is decided once, at registration, from three properties of the dependency: whether there is map-side aggregation, how many partitions there are, and whether the serializer can relocate serialized records. The first two you control directly. The third is chosen for you from your record types, and it is the one worth checking when a shuffle costs more than it should.

References

• Spark shuffle configuration for every spark.shuffle.* setting and its default
• Spark memory management for the execution pool a sorter spills against
• Data skew in Apache Spark for the failure mode that looks like a shuffle problem and is not

Trademarks

Apache Spark, Apache and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries.