How do I resolve a shuffle fetch failure?

Summary

• To resolve a shuffle fetch failure, find out what happened to the executor that wrote the missing shuffle blocks; the task reporting the error was only reading that executor's output.
• Spark reacts by resubmitting the stage that produced the lost map output and aborts after spark.stage.maxConsecutiveAttempts (default 4) consecutive attempts. Fetch failures caused by executor decommission don't count toward that limit by default.
• "Missing an output location for shuffle" comes from MetadataFetchFailedException: Spark's map output tracker has no registered location for that map output, which points to output lost along with its executor.
• Match the fix to the finding: memory and skew changes for an executor that died, shuffle tracking, an external shuffle service or decommissioning for one that was removed, and retry or fetch-concurrency limits for one that stalled.
• Confirm on the rerun: no retried attempts for the stage in the Stages tab, and a low Shuffle Read Fetch Wait Time.

How do I resolve a shuffle fetch failure?

Treat the error as a pointer to a different executor. It tells you a reduce task couldn't get map output that another executor wrote earlier, so the fix depends on what happened to that executor: it died, it was removed, or it was too busy to answer. Work out which one, make the matching change, and confirm it on the rerun. Raising retry counts without that diagnosis mostly delays the same failure.

What Spark does when a fetch fails

In a shuffle, each executor "first writes its own map outputs locally to disk, and then acts as the server for those files when other executors attempt to fetch them." When a reducer can't fetch a block, its executor reports a FetchFailedException to the scheduler, which resubmits the earlier stage to rebuild the lost output. Spark aborts after spark.stage.maxConsecutiveAttempts consecutive stage attempts (default 4). Because spark.stage.ignoreDecommissionFetchFailure defaults to true, fetch failures caused by executor decommission aren't counted toward that limit.

Step 1: Read the failure

Start in the Spark UI's Stages tab, where "only failed stages show the failure reason." For a fetch failure, the reason identifies the shuffle, the map output and the block manager, meaning the executor, that should have served it. When there are many attempts, the monitoring REST API returns failed stages as JSON:

curl "http://localhost:4040/api/v1/applications/<app-id>/stages?status=failed"

Tell two messages apart. A FetchFailedException that names a block manager means the reducer reached for a specific executor and couldn't read from it. MetadataFetchFailedException with "Missing an output location for shuffle" means Spark's map output tracker has no live location for that output, for example after the executor that wrote it was lost.

Step 2: Find out what happened to the serving executor

Open the Executors tab, which shows each executor's memory, disk, shuffle totals and GC time, and read that executor's stderr log. Look for one of four stories:

• It died. Out-of-memory errors or a container kill shortly before the failure.
• It was removed. Dynamic allocation scaled it down, it was decommissioned, or its spot or preemptible node was reclaimed.
• It stalled. High GC time, or a node serving many reducers at once.
• It ran out of disk. Shuffle files live in spark.local.dir (default /tmp, often overridden by the cluster manager), which the docs say "should be on a fast, local disk."

Step 3: Apply the fix that matches

Option	When to use it	Trade-off
Raise spark.executor.memoryOverhead (YARN and Kubernetes), or use more and smaller shuffle partitions	The serving executor was killed for memory	More memory per executor, or more tasks
Keep AQE skew handling on (spark.sql.adaptive.skewJoin.enabled, default true)	A few oversized partitions dominate a sort-merge join	Needs AQE (spark.sql.adaptive.enabled) to stay on
Shuffle tracking, the external shuffle service (spark.shuffle.service.enabled) or decommissioning with shuffle block migration	Executors leave while their shuffle data is still needed	The shuffle service is a separate long-running process on every node
Lower spark.network.maxRemoteBlockSizeFetchToMem (default 200m)	Large remote blocks put memory pressure on reducers	Blocks above the threshold are fetched to disk
Lower spark.reducer.maxReqsInFlight or spark.reducer.maxBlocksInFlightPerAddress	One node is overwhelmed by concurrent fetch requests	Less read parallelism
Increase spark.shuffle.io.maxRetries (3 by default) and spark.shuffle.io.retryWait (5s by default)	Long GC pauses or transient network issues	A real failure takes longer to surface

Pass these when the application launches: Spark's configuration guide notes that deployment properties set programmatically at runtime may have no effect. The values below are examples, not recommendations:

from pyspark.sql import SparkSession

spark = (SparkSession.builder
         .config("spark.executor.memoryOverhead", "3g")
         .config("spark.sql.shuffle.partitions", "800")
         .config("spark.network.maxRemoteBlockSizeFetchToMem", "64m")
         .config("spark.shuffle.io.maxRetries", "8")
         .getOrCreate())

Step 4: Confirm on the rerun

On the rerun, the failed stage should complete without retried attempts, and its Shuffle Read Fetch Wait Time, "the time that tasks spent blocked waiting for shuffle data to be read from remote machines," should stay low. If failures jump to a different executor each time, return to Step 2: something is still killing or removing executors.

How Databricks approaches fetch failures

On Databricks, audit the cluster's Spark configuration first. Databricks advises against setting most Spark properties, since overrides carried over from open-source Spark can mask newer runtime defaults. Shuffle sizing can be left to the platform: the value auto for spark.sql.shuffle.partitions enables auto-optimized shuffle, where Databricks derives the partition count from the plan and the size of the input. Serverless compute uses auto by default and accepts only a short list of Spark properties, because Databricks manages that compute for you.

FAQs

Should I just rerun the job?

A rerun can succeed when the cause was transient, such as a GC pause or a network blip. If an executor was killed for memory or removed mid-shuffle, expect the same failure until you change the configuration.

Does the external shuffle service prevent fetch failures?

It covers the ones caused by losing an executor: the docs say it preserves shuffle files "so that shuffle fetches can continue in the event of executor failure." It can't help when the whole node disappears, because the service runs on that node.