What is a shuffle in Spark or a distributed query engine?

A shuffle is the all-to-all move that re-sends every row to a machine chosen by a hash of its key, so rows that must be compared end up together. Spark defines it as "Spark's mechanism for re-distributing data so that it's grouped differently across partitions." It is the step that makes a distributed join or group-by possible, because a machine can only compare rows that are in its own memory.

Why is the shuffle the slowest part of a query?

Because it touches all three slow resources at once. Spark: "The Shuffle is an expensive operation since it involves disk I/O, data serialization, and network I/O." Every other operator (filter, project, hash lookup) runs on data a machine already holds, at memory speed; the shuffle serializes rows to bytes, writes map outputs to disk, and moves them across the network — for potentially every row. In real analytics queries the wall clock is dominated by shuffles, not computation.

What is data skew and why does one hot key create a straggler?

Skew is a distribution where one or a few keys carry far more rows than the rest — a mega-customer, a NULL key, a viral post. Since the exchange routes by hash(key) mod N, every row for a hot key hashes to the same partition and lands on one reduce task. That task holds a mountain of rows and spills to disk while its neighbours idle. The whole query waits on its slowest task — the straggler — so a job can have 199 idle machines waiting on one. Adding machines does not help: the hot key still hashes to one partition.

How does Spark AQE fix skew, and what are the defaults?

Adaptive Query Execution measures real partition sizes after the shuffle, then "dynamically handles skew in sort-merge join by splitting (and replicating if needed) skewed partitions into roughly evenly sized tasks." A partition counts as skewed when it exceeds 5× the median (skewedPartitionFactor = 5.0) AND 256 MB (skewedPartitionThresholdInBytes); it is split into ~64 MB tasks (advisoryPartitionSizeInBytes) run in parallel. AQE is on by default since Spark 3.2.

What is a broadcast join and when does it beat a shuffle?

A broadcast join ships a full copy of the small side of a join to every worker, so each joins its local slice of the big table against the in-memory copy — no rows of the big table ever move, so there is no shuffle and skew is harmless. Spark broadcasts automatically when a side fits under spark.sql.autoBroadcastJoinThreshold, default 10 MB. It is the escape hatch: the cheapest shuffle is the one you never run.

The Shuffle: How a Cluster Moves a Join

A join is trivial once the matching rows share a machine. The shuffle is the all-to-all network move that gets them there — repartition every row by a hash of its key, and everything hard follows: why it dominates query time, why one hot key strands a straggler while the fleet idles, and how adaptive splits, salting, and broadcast joins break it.

Concept · Systems. The source ↗

A free, interactive, animated visual explainer of The Shuffle: How a Cluster Moves a Join — built to be understood, not skimmed.

Questions

What is a shuffle in Spark or a distributed query engine?: A shuffle is the all-to-all move that re-sends every row to a machine chosen by a hash of its key, so rows that must be compared end up together. Spark defines it as "Spark's mechanism for re-distributing data so that it's grouped differently across partitions." It is the step that makes a distributed join or group-by possible, because a machine can only compare rows that are in its own memory.
Why is the shuffle the slowest part of a query?: Because it touches all three slow resources at once. Spark: "The Shuffle is an expensive operation since it involves disk I/O, data serialization, and network I/O." Every other operator (filter, project, hash lookup) runs on data a machine already holds, at memory speed; the shuffle serializes rows to bytes, writes map outputs to disk, and moves them across the network — for potentially every row. In real analytics queries the wall clock is dominated by shuffles, not computation.
What is data skew and why does one hot key create a straggler?: Skew is a distribution where one or a few keys carry far more rows than the rest — a mega-customer, a NULL key, a viral post. Since the exchange routes by hash(key) mod N, every row for a hot key hashes to the same partition and lands on one reduce task. That task holds a mountain of rows and spills to disk while its neighbours idle. The whole query waits on its slowest task — the straggler — so a job can have 199 idle machines waiting on one. Adding machines does not help: the hot key still hashes to one partition.
How does Spark AQE fix skew, and what are the defaults?: Adaptive Query Execution measures real partition sizes after the shuffle, then "dynamically handles skew in sort-merge join by splitting (and replicating if needed) skewed partitions into roughly evenly sized tasks." A partition counts as skewed when it exceeds 5× the median (skewedPartitionFactor = 5.0) AND 256 MB (skewedPartitionThresholdInBytes); it is split into ~64 MB tasks (advisoryPartitionSizeInBytes) run in parallel. AQE is on by default since Spark 3.2.
What is a broadcast join and when does it beat a shuffle?: A broadcast join ships a full copy of the small side of a join to every worker, so each joins its local slice of the big table against the in-memory copy — no rows of the big table ever move, so there is no shuffle and skew is harmless. Spark broadcasts automatically when a side fits under spark.sql.autoBroadcastJoinThreshold, default 10 MB. It is the escape hatch: the cheapest shuffle is the one you never run.

The Shuffle: How a Cluster Moves a Join

Questions

Related explainers