The Shuffle Tuning Knobs

With only the needed shuffles left in the plan, configuration sets how each remaining one runs. The knobs sort by the part of the shuffle they touch, and each one answers a symptom you can see on the stage page.

Knob	Default	What it changes
`spark.sql.shuffle.partitions`	200	reducer count R, so block count and bytes per reducer
`spark.shuffle.sort.bypassMergeThreshold`	200	largest R that uses the bypass writer
`spark.shuffle.file.buffer`	32KB	memory buffer in front of each file a map task writes
`spark.shuffle.compress`	true	compression of map output
`spark.reducer.maxSizeInFlight`	48MB	bytes a reduce task keeps requested at once
`spark.reducer.maxBlocksInFlightPerAddress`	no limit	blocks one reducer requests from 1 server at once
`spark.shuffle.io.maxRetries`	3	fetch retries before a fetch failure
`spark.shuffle.io.retryWait`	5s	wait between those retries
`spark.shuffle.service.enabled`	false	serve files from the node's shuffle service
`spark.shuffle.push.enabled`	false	push and merge blocks at remote services
`spark.local.dir`	/tmp	directories for shuffle and spill files

Map side: larger, fewer disk writes

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.

Reduce side: how hard reducers pull

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.

Fetch retries: how long before a rerun

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.

Local directories for shuffle files

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.

spark = (SparkSession.builder
    .appName("daily_clicks")
    .config("spark.shuffle.file.buffer", "1m")
    .config("spark.reducer.maxSizeInFlight", "96m")
    .config("spark.shuffle.io.maxRetries", "10")
    .config("spark.shuffle.io.retryWait", "10s")
    .config("spark.shuffle.service.enabled", "true")
    .getOrCreate())

Do

Delete unneeded shuffles before tuning the ones that remain
Change 1 knob per run and compare the same stage's metrics
Check Shuffle Read Blocked Time before raising maxSizeInFlight
Put spark.local.dir on fast local disks, 1 directory per disk

Avoid

Raise spark.shuffle.file.buffer without counting buffers per task
Raise fetch retries to hide a node that keeps failing
Enable the shuffle service before it runs on every node
Judge a change by total job time alone

Reduce tasks spend most of their time in Shuffle Read Blocked Time. The executors have memory to spare, and the shuffle services report light load. Which change targets that symptom first?

Raise spark.shuffle.sort.bypassMergeThreshold so map tasks write fewer files
Raise spark.shuffle.io.maxRetries so slow fetches get more attempts
Raise spark.reducer.maxSizeInFlight so each reducer keeps more bytes requested
Lower spark.reducer.maxBlocksInFlightPerAddress so servers get fewer requests

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.