Shuffles: Advanced

Since Spark 2.0, the sort-based shuffle manager is the only one Spark ships. A map task with output for R reducers writes 2 files. The data file holds all of its output rows as R segments laid end to end in reducer order. The index file holds R + 1 byte offsets: where each segment starts, plus the end of the file. Reducer 3 reads from offset 3 to offset 4, and those bytes are its block from this map task.

DataFrame aggregations do their combining in the partial aggregate before the shuffle, so their shuffle writer never combines. That leaves R as the deciding number. With spark.sql.shuffle.partitions at its default of 200, R equals the threshold, and a default DataFrame shuffle takes the bypass path: each map task holds 200 files open while it writes. Set 201 or more and the same job switches to the serialized writer, with no change to its output.

Before fetching, a reduce task asks the driver where every map task's output lives and how big each block is. It then pulls its block from each map task: from local disk when the map ran on its own executor, over the network otherwise. spark.reducer.maxSizeInFlight, 48MB by default, limits how many bytes it has requested and not yet received. If the operation needs key order, the reducer sorts what arrives and spills when it runs short of memory.

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.

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.

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.

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.

Shuffle output is compressed by default: spark.shuffle.compress is true and the codec is lz4. Shuffle Write Size therefore reports compressed bytes, the same bytes that later cross the disk and the network.

The cell below reads 3 counts that bound a shuffle of the seed's orders: rows read, rows left after keeping completed orders, and the distinct regions among them. For a groupBy("region") on completed orders, that last count caps the partial sums each map task can send: at most 1 per distinct region it holds, plus 1 for rows whose region is null, since countDistinct skips nulls. Run it, then change "Completed" to another status and see which counts move.

orders = spark.table("orders")

completed = F.col("status") == "Completed"

bounds = orders.agg(
    F.count("*").alias("rows_read"),
    F.sum(F.when(completed, 1).otherwise(0)).alias("rows_after_filter"),
    F.countDistinct(F.when(completed, F.col("region"))).alias("regions_after_filter"),
)

bounds.show()

Spark records how each DataFrame is partitioned. After repartition("customer_id"), every row for a customer sits in 1 partition, so a groupBy("customer_id") that follows finds its requirement met and adds no shuffle of its own. The board plays exactly that: 2 wide operations in the code, rows exchanged once.

A hash layout on user_id also satisfies a grouping on user_id plus other columns, since rows that share a (user_id, event_date) pair already share a user_id. The job below counts each user's events per day, then keeps each user's busiest day.

The drill hands you a program that already repartitions the seed's order_items by order_id. You choose the grouping columns and aggregates for both steps so the result is each order's largest per-product quantity, with both aggregations reusing that single shuffle.

The bypass writer keeps 1 file buffer per reducer, so at 200 reducers a map task holds around 6.4MB of buffers. Raising spark.shuffle.file.buffer to 1MB makes about 32 times fewer write calls but holds 200MB per task. Raise it when Shuffle Write Time is a large share of map task time and executors have memory to spare. Leave spark.shuffle.sort.bypassMergeThreshold alone unless you have measured both writers on your job, since raising it means more files open at once in every map task.

When Shuffle Read Blocked Time dominates the reduce stage and the serving nodes are lightly loaded, raising spark.reducer.maxSizeInFlight to 96MB keeps more requests outstanding, so transfers overlap with compute. spark.reducer.maxBlocksInFlightPerAddress works the other way: lowering it caps how many blocks 1 reducer asks of 1 server at a time, which protects shuffle services that drop connections under a flood of small requests.

A failed fetch waits spark.shuffle.io.retryWait and tries again, up to spark.shuffle.io.maxRetries times, and how long that takes depends on how the server fails. A server that refuses connections fails each attempt at once, so the defaults of 3 retries 5 seconds apart turn it into a fetch failure and a rerun of map tasks after about 15 seconds. At 10 retries 10 seconds apart, that stretches to about 100 seconds. A server that hangs makes each attempt first wait out spark.shuffle.io.connectionTimeout, which falls back to spark.network.timeout, 120 seconds by default. With the defaults that is 4 attempts of up to 120 seconds plus 15 seconds of waits, about 8 minutes, and at 10 retries 10 seconds apart about 24 minutes. More patience rides out shuffle servers that pause under heavy load, and every second of it is also how long a dead node takes to be declared lost.

spark.local.dir names the directories that receive shuffle and spill files, and cluster managers often override it with their own setting. List 1 directory per local disk, separated by commas, and Spark spreads files across them. Shuffle files on a network-attached volume pay the network twice: once to write and once to serve.

Most of these settings are read when executors start, so they belong in the submit-time configuration. spark.sql.shuffle.partitions is the exception: spark.conf.set changes it between queries.