A Spark stage fails with FetchFailedException — what happened, and what does Spark do next?

## What the exception actually means `FetchFailedException` is raised by the shuffle reader, but the failure lives upstream. The reduce task asked the map output tracker where its partition's blocks are, got an address, and could not get the bytes: connection refused, connection reset, a timeout, or a corrupt/missing block. Spark retried the fetch `spark.shuffle.io.maxRetries` times (default 3, with a wait between attempts) before giving up. The stack trace names the host and port it failed to reach — port 7337 in the message points at the external shuffle service, another port at a peer executor's own block-transfer server. ## Why the map output was gone Walk the causes in the order that finds most incidents: 1. **The executor died.** JVM OOM, a container killed by YARN for exceeding physical memory (the classic "Container killed by YARN for exceeding memory limits" message), or a Kubernetes pod evicted. Without an external shuffle service, the moment the JVM exits its shuffle files become unreachable even though they still exist on disk — nothing is left to serve them. 2. **The node went away.** Spot or preemptible instance reclaimed, node drained, hardware failure. The files really are gone. 3. **Local disk filled.** Shuffle-heavy jobs write a lot under `spark.local.dir`; a full disk fails the write or truncates it, and the reader sees a missing or short block. 4. **The server was alive but unresponsive.** A multi-second stop-the-world GC on the serving executor, or an external shuffle service overwhelmed by thousands of concurrent fetch requests, blows past `spark.network.timeout` (default 120s). 5. **Very large blocks.** A skewed partition can put a huge block behind one fetch; slow transfers and memory pressure on both ends make timeouts likelier. ## What the scheduler does about it This is the part interviewers actually want. A fetch failure is not an ordinary task failure. The `DAGScheduler` receives it and: - unregisters all map outputs held by that executor (and, if the whole host is suspected lost, by every executor on it), so the tracker stops handing out dead addresses; - marks the corresponding **map stage** as having missing partitions and resubmits it — only the missing map tasks rerun, not the whole stage's worth of work if other outputs survive; - resubmits the reduce stage afterwards. Crucially it does **not** count toward `spark.task.maxFailures`, because rerunning the reader would fail identically; the missing input must be regenerated first. What does bound it is `spark.stage.maxConsecutiveAttempts` (default 4): after that many consecutive failed attempts at the stage, the job aborts. That is why a cluster losing executors steadily produces the familiar pattern of a job crawling through repeated stage retries and then dying — each retry recomputes expensive upstream work, so wall-clock cost is superlinear. ## How to diagnose it in practice Do not start in the reduce stage's logs. Take the host named in the exception and look at **that executor's** log and its exit reason. Three readings: - Executor log ends with an OOM or the resource manager's kill message → a memory problem: raise `spark.executor.memoryOverhead` or executor memory, reduce per-task memory demand by increasing shuffle partitions, or lower cores per executor so fewer tasks share the heap. - Executor still alive, failure is a timeout or connection reset → serving-side congestion or GC. Look at the shuffle service logs, block-transfer thread settings and GC pauses on that node. - Whole node absent from the cluster → infrastructure: preemption, autoscaling, hardware. ## How to stop it recurring The durable fixes are structural, not retry-count tuning. Run an **external shuffle service** so files outlive the executor process, or on deployments without one, enable graceful decommissioning so shuffle blocks migrate before an executor goes away. Right-size executors so they are not killed for overhead. Increase shuffle partitions so individual blocks are smaller and no single fetch is enormous. Keep spare local disk. Raising `spark.shuffle.io.maxRetries` and `spark.network.timeout` buys tolerance for transient congestion and is a legitimate mitigation — but it is a band-aid over dying executors, and saying "just increase the retries" is the answer that ends interviews. ## What interviewers are checking That you locate the cause upstream of where the error appeared, that you know the scheduler regenerates map output rather than simply retrying the failed task, and that you can name the bounded retry budget that turns this into a job failure.