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

• Why does a shuffle exist?
• 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 happens when a fetch fails?
• Which numbers actually tell you something?
• 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
• 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.
• 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.

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 keeps a map when aggregating and a buffer when not, spills sorted runs to disk under memory pressure, and merges them at the end.

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 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.

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 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.

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.

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. Sort-based shuffle trades a sort for a file count of M, and 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.

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
• SortShuffleManager, where registerShuffle makes the choice this post is about
• SerializerManager, which auto-picks Kryo from the key and value class tags
• UnsafeShuffleWriter for the serialized sort and the transferTo merge
• 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.