What is a shuffle, and why is it expensive?

Summary

• A shuffle is Spark's mechanism for re-distributing data so that it's grouped differently across partitions. The Spark docs say it "typically involves copying data across executors and machines, making the shuffle a complex and costly operation."
• Operations that can cause a shuffle include repartition and coalesce, ByKey operations such as groupByKey and reduceByKey, and joins such as cogroup and join.
• The cost comes from disk I/O, data serialization and network I/O, plus in-memory data structures that can spill to disk and intermediate files that stay on disk until garbage collection.
• On Databricks, Photon "uses a redesigned columnar shuffle to increase throughput for large-scale joins."

What is a shuffle, and why is it expensive?

"The shuffle is Spark's mechanism for re-distributing data so that it's grouped differently across partitions." Take reduceByKey: all values for one key must be in the same place to compute the result, but they may sit on different partitions or machines. Spark has to perform an all-to-all operation: read from all partitions to find the values for every key, then bring values together across partitions to compute each key's result. That step is the shuffle.

Which operations cause a shuffle?

The RDD programming guide names three families: repartition operations such as repartition and coalesce, ByKey operations (except for counting) such as groupByKey and reduceByKey, and join operations such as cogroup and join. The API docs add detail: RDD repartition "uses a shuffle to redistribute data", while RDD coalesce takes a shuffle flag (default False) for "whether to add a shuffle step".

Why is a shuffle expensive?

Cost	What the Spark docs say
Disk I/O	Map task results stay in memory until they can't fit, then are sorted by target partition and written to a single file. Reduce tasks read the relevant sorted blocks
Serialization and network I/O	The shuffle "involves disk I/O, data serialization, and network I/O"
Memory and spill	reduceByKey and aggregateByKey build in-memory data structures on the map side, and ByKey operations build them on the reduce side. When data does not fit, Spark spills them to disk, adding disk I/O and garbage collection
Intermediate files	Shuffle files are kept until the RDDs are garbage collected, so long-running jobs can use a large amount of disk space

How do I see a shuffle in the Spark UI?

Every SparkContext launches a Web UI, by default on port 4040 (http://<driver-node>:4040). Open a stage from the Stages tab. The stage DAG labels nodes by operation scope, such as Exchange, and the summary metrics include:

Metric	Definition from the Spark docs
Shuffle Read Size / Records	Total shuffle bytes read, from local and remote executors
Shuffle Remote Reads	Total shuffle bytes read from remote executors
Shuffle Read Fetch Wait Time	Time tasks spent blocked waiting for shuffle data from remote machines
Shuffle Write Time	Time tasks spent writing shuffle data
Shuffle spill (memory)	Size of the deserialized form of the shuffled data in memory
Shuffle spill (disk)	Size of the serialized form of the data on disk

How do I reduce shuffle cost?

• Aggregate with reduceByKey or aggregateByKey. The groupByKey docs say that for a sum or average per key these "will provide much better performance". reduceByKey also merges locally on each mapper before sending results to a reducer.

from operator import add
rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
sorted(rdd.groupByKey().mapValues(sum).collect())  # groups all values first
sorted(rdd.reduceByKey(add).collect())             # merges locally first

• Reduce partitions with coalesce. The DataFrame docs say going from many partitions to fewer "results in a narrow dependency" with no shuffle, but a drastic coalesce, such as to one partition, may run on fewer nodes than you like. Calling repartition() instead "will add a shuffle step" but keeps upstream partitions running in parallel.
• Sort while repartitioning. repartitionAndSortWithinPartitions can push the sorting down into the shuffle machinery.
• Raise parallelism if reduce tasks run out of memory. Shuffle operations "build a hash table within each task", and the tuning guide's simplest fix is to increase the level of parallelism so each task's input is smaller.
• Persist an RDD you will reuse. The guide recommends calling persist on the resulting RDD.

Which settings affect shuffles?

Setting	Default	What the Spark docs say
spark.shuffle.compress	true	Whether to compress map output files
spark.shuffle.spill.compress	true	Whether to compress data spilled during shuffles
spark.reducer.maxSizeInFlight	48m	Maximum size of map outputs to fetch simultaneously from each reduce task, a fixed memory overhead per reduce task

How does Databricks handle shuffles?

Photon is "the Databricks-native vectorized query engine." Its architecture "uses a redesigned columnar shuffle to increase throughput for large-scale joins." The Photon docs also say it doesn't support RDD APIs, so Photon does not run the RDD examples above.
For Spark SQL, spark.sql.shuffle.partitions defaults to 200 on Databricks. Setting it to auto "enables auto-optimized shuffle, which automatically determines this number based on the query plan and the query input data size." The Databricks Spark UI guide recommends spark.sql.shuffle.partitions=auto for stages with high shuffle.

FAQs

Which setting controls the number of shuffle partitions?

For Spark SQL, spark.sql.shuffle.partitions sets the number of partitions used when shuffling data for joins or aggregations (default 200). For RDD operations, pass numPartitions or set spark.default.parallelism.

How long do shuffle files stay on disk?

The guide says they are preserved until the corresponding RDDs are no longer used and are garbage collected, so they don't need to be re-created if the lineage is re-computed. The scratch directory is spark.local.dir.