How do I fix shuffle fetch failures?

Summary

• A shuffle fetch failure means a reduce task couldn't read a block of map output from another executor. Spark throws FetchFailedException, retries the previous stage to rebuild the missing output, and aborts the stage after spark.stage.maxConsecutiveAttempts consecutive attempts (default 4).
• Treat the error as a symptom. The executor that held the shuffle files was killed, removed, overloaded or briefly unreachable while a reducer was reading from it.
• Fix the cause before touching retries: give executors enough memory headroom, keep shuffle files available when executors leave (external shuffle service, shuffle tracking or decommissioning), and split skewed partitions with adaptive query execution (AQE).
• Retry settings such as spark.shuffle.io.maxRetries and spark.shuffle.io.retryWait help with transient problems like long GC pauses or network blips. They can't bring back the files of an executor that is gone.

How do I fix shuffle fetch failures?

Find out why the executor that wrote the shuffle data stopped serving it, fix that, and raise retry or timeout settings last. A fetch failure means a reduce task couldn't read map output from another executor, so Spark throws FetchFailedException and retries the previous stage to regenerate that output. If the same stage keeps failing, Spark aborts it after spark.stage.maxConsecutiveAttempts consecutive attempts (default 4). The configuration names and defaults below are from the Apache Spark 4.2.0 documentation.

What is a shuffle fetch failure?

During a shuffle, map tasks sort their output by target partition and write it to files on local disk. Reduce tasks then read the blocks they need, many of them from other executors over the network. Those files stay with the executor that wrote them, or with the external shuffle service if you run one. When that process dies, gets removed, or can't answer in time, the reducer's fetch fails and Spark has to recompute the map output.
Two places in the Spark UI narrow it down. On the stage page, Shuffle Read Fetch Wait Time shows how long tasks were blocked waiting for remote shuffle data, and Shuffle Remote Reads shows the bytes read from other executors. The Executors tab lists every executor allocated to the application, with GC time. Match the failure's timestamp against executor exits in your cluster manager's logs.

Diagnose the cause, then apply the matching fix

Symptom	Likely cause	Fix
An executor exits or its container is killed just before the failure	The executor outgrew its container, which is sized as spark.executor.memory plus spark.executor.memoryOverhead, spark.memory.offHeap.size and spark.executor.pyspark.memory	Raise spark.executor.memoryOverhead or executor memory, or shrink per-task data with more shuffle partitions
Failures follow node preemption, scale-down or dynamic allocation removing executors	Shuffle files left with the executor	Run the external shuffle service (spark.shuffle.service.enabled, default false), keep spark.dynamicAllocation.shuffleTracking.enabled (default true), or enable decommissioning with shuffle-block migration
Fetch timeouts during long GC pauses or brief network drops	The serving executor stalled	Reduce GC pressure first, then raise spark.shuffle.io.maxRetries (default 3) or spark.shuffle.io.retryWait (default 5s), and review spark.network.timeout (default 120s)
A stage keeps failing on a few oversized partitions	Skewed or oversized shuffle blocks	Keep spark.sql.adaptive.enabled and spark.sql.adaptive.skewJoin.enabled on (both default true), or raise spark.sql.shuffle.partitions (default 200)
A node or its shuffle service falls over when many reducers read at once	Too many concurrent fetch requests	Lower spark.reducer.maxReqsInFlight or spark.reducer.maxBlocksInFlightPerAddress (both default Int.MaxValue)

The Spark docs suggest setting deploy-related properties such as spark.executor.instances through spark-submit or the configuration file, because setting them at runtime may have no effect. The values here are examples; size them for your executors and cluster.

spark-submit \
  --conf spark.executor.memoryOverhead=2g \
  --conf spark.shuffle.io.maxRetries=6 \
  --conf spark.shuffle.io.retryWait=10s \
  --conf spark.reducer.maxBlocksInFlightPerAddress=64 \
  my_job.py

If executors are removed on purpose (spot capacity, autoscaling, Kubernetes scale-down), let Spark move their shuffle blocks before they shut down:

spark.decommission.enabled                        true
spark.storage.decommission.enabled                true
spark.storage.decommission.shuffleBlocks.enabled  true
spark.storage.decommission.fallbackStorage.path   s3a://<bucket>/spark-fallback/

These lines go in conf/spark-defaults.conf. The last one is optional: it gives blocks somewhere to go when no peer executor can take them. Spark doesn't clean up that path, so put a lifecycle or TTL policy on it.

What changed in Spark 4.x

Recent releases harden how Spark recovers. The Spark 4.1.0 release notes list "Fast fail when shuffle fetch failure happens" (SPARK-52395) and a checksum-based full retry of shuffle stages to avoid incorrect results (SPARK-51756). Spark 4.2.0 rolls back and fully retries succeeding shuffle map stages when a shuffle checksum mismatch is detected, instead of consuming inconsistent data (SPARK-54556). Upgrading changes the recovery path. It doesn't stop executors from dying, so the fixes above still apply.

How Databricks approaches shuffle fetch failures

On Databricks classic compute, the docs name shuffle fetch failures among the problems that spot-instance preemption causes, say you can enable decommissioning to help, and recommend enabling data migration when you do, with the same spark.decommission.enabled, spark.storage.decommission.enabled and spark.storage.decommission.shuffleBlocks.enabled properties. They also say decommissioning is best effort: a task already fetching from a preempted executor can still fail. Serverless compute doesn't expose the Spark UI, so use the query profile instead. See Decommission spot instances.

FAQs

Will raising spark.shuffle.io.maxRetries fix it?

Only when the cause is transient. The Spark docs say retries help large shuffles survive long GC pauses or transient network problems, and the maximum delay from retrying is maxRetries * retryWait, which is 15 seconds with the defaults. If the executor is gone, so are its shuffle files.

Can I use the external shuffle service on Kubernetes?

No. The Spark job-scheduling guide says Spark on Kubernetes doesn't yet support the external shuffle service. Use shuffle tracking, which is on by default, or decommissioning with shuffle-block migration.

Is shuffle corruption the same problem?

It's related. spark.shuffle.detectCorrupt (default true) checks fetched blocks for corruption, and spark.shuffle.checksum.enabled (default true) stores checksums so Spark can work out whether a network or disk issue caused it.

Does adding shuffle partitions help?

It can, when tasks fail while fetching oversized blocks. Smaller partitions mean smaller blocks per fetch, and AQE can coalesce contiguous partitions afterward if they end up too small.