The External Shuffle Service

The data and index files from the map stage land under spark.local.dir on the node where each map task ran. Unless you configure otherwise, the executor that wrote them also serves them: its block manager answers every reducer's request for a byte range. The files can be read only while that executor is alive.

Losing the executor that wrote the files

Executors leave mid-job for ordinary reasons: killed for exceeding their memory limit, preempted by the cluster manager, or released when idle. Suppose executor 7 ran 40 map tasks and then dies. Reducers asking it for blocks get a fetch failure, which Spark handles differently from the task failures you have met. Retrying the reduce task cannot help, because the bytes are gone. The DAGScheduler marks executor 7's map outputs as lost, resubmits the map stage to rerun those 40 tasks, and then reruns the reduce tasks that hadn't finished. In the Stages tab, the rerun appears with (retry 1) beside the stage id.

A server that outlives executors

The external shuffle service is a long-running process on every worker node, started by the cluster manager outside any application and shared by all of them. When an executor starts, it registers its local directories with the service on its node. Reducers then fetch blocks from the service, on port 7337 by default, and the service reads the bytes off disk itself. An executor can exit at any point after its map tasks finish, and its files stay readable until the application ends.

The map task writes its data and index files to node A's disk. The reduce task learns block locations from the driver, then fetches its byte ranges from node A's shuffle service, which keeps serving after executor 7 exits.

Map task (transform, executor 7, node A)
Local disk (storage, node A, spark.local.dir)
Shuffle service (custom, node A, port 7337)
Driver (custom, map output locations)
Reduce task (consumer, executor 22, node B)

Map task -> Local disk
Local disk -> Shuffle service
Shuffle service -> Reduce task
Driver -> Reduce task

spark.shuffle.service.enabled: true
spark.shuffle.service.port: 7337

Executor serves its files

Files readable only while the writing executor runs
A lost executor reruns its map tasks
Serving shares the executor's threads with its running tasks
Nothing extra to install on the nodes

Shuffle service serves them

Files readable while the node stays up
A lost executor costs no rerun, and fetches continue
Serving runs in its own process on the node
The service must run, and stay compatible, on every node

A lost node still costs a rerun

The service reads files from a local disk and copies them nowhere. If the node goes down or its disk fails, every map output on that node is gone. If only the service process dies, the files still sit on disk but nothing can serve them: on the first fetch failure Spark unregisters that host's map outputs, and the DAGScheduler reruns those map tasks as it would without the service. The service turns an executor loss into a non-event, while a node loss still costs a partial rerun of the map stage.

Push-based shuffle

The service can also fix the block count that the sort layout leaves behind. A shuffle from 10,000 map tasks to 10,000 reducers stores 100 million blocks, so a 1TB shuffle averages 10KB per block. Each reducer issues 10,000 small reads scattered across the cluster's disks, and on spinning disks the seeks cost more than the bytes. LinkedIn's engineers described this in their Magnet paper and contributed push-based shuffle to Spark 3.2.

With spark.shuffle.push.enabled set to true, a map task writes its files as usual, then pushes its blocks to shuffle services Spark assigned as mergers, each responsible for a set of reducers. A merger appends every block for a reducer into 1 merged file. The reducer reads a few large merged chunks instead of 10,000 small blocks, and falls back to the original files for any block that missed the merge. It requires the external shuffle service, and not every cluster manager supports it.

An application runs with the external shuffle service enabled. Halfway through a reduce stage, executor 12 is killed for exceeding its memory limit, and its node stays up. What happens to the shuffle files executor 12 wrote?

Reducers keep fetching them from the node's shuffle service, and no map task reruns
They are deleted with the executor, so the map stage reruns executor 12's tasks
The service keeps serving them, but the DAGScheduler reruns executor 12's tasks anyway
Reducers fetch them from a copy the service placed on another node at write time

About this lesson

This section is part of the Shuffles: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.