What is a shuffle in Spark, and how do you read it in the Spark UI?

Most of a Spark job happens inside partitions: each task reads its slice of the data, filters it, adds columns, and no row leaves the executor it started on. A shuffle is the exception. To join on customer_id or add up sales by country, every row with the same key must be in the same partition, so Spark redistributes the whole dataset. It is the slowest thing a job usually does, and the place where skewShuffle and Data SkewA shuffle moves rows between machines so rows with the same key meet, for a join or group by; skew is when one key has far more rows, so one task does most of the work. and spillSpark SpillWhen a task's sort, join or aggregation no longer fits in execution memory, Spark writes part of it to local disk and reads it back later, so the task finishes, only slower. show up.

The 2 sides

A shuffle is a boundary between 2 stages.

1. Map side, the shuffle write. Each task of the first stage works out which partition every row belongs to, by hashing the key. It holds rows in memory, sorts them by target partition and writes them to one file on local disk.
2. Reduce side, the shuffle read. Each task of the next stage takes one target partition and fetches its block from every map output file, some on its own executor (local bytes) and most on others (remote bytes).

So a shuffle costs disk writes, turning rows into bytes, and network traffic. The files stay on disk while the data might be needed again, which is why dynamic allocation needs an external shuffle service or shuffle tracking before it can remove an executor.

How to spot one

• In a plan, a shuffle is an Exchange. Exchange hashpartitioning(customer_id, 200) means: hash on customer_id into 200 partitions. explain() prints it, and the SQL / DataFrame tab draws it as a box.
• In the Jobs tab, a job with 2 or more stages has a shuffle between each pair.
• On a stage page, Shuffle Write on the first stage and Shuffle Read on the next are the same data seen from both ends. Compare the max shuffle read per task with the median: a max many times the median is skew. A high fetch wait time means tasks sat waiting for blocks over the network.

How many partitions

After a shuffle, the data is split into spark.sql.shuffle.partitions partitions, 200 unless you set it. 200 is too many for a few megabytes and too few for a terabyte. Adaptive query executionAdaptive Query ExecutionSpark's way of re-planning a query while it runs: after each shuffle it looks at the real sizes, then merges small partitions, splits skewed ones and can switch a join to a broadcast. fixes the first case for you by merging small partitions after it sees their real size; the second case needs a higher setting, or partitions that come out too big and spillSpark SpillWhen a task's sort, join or aggregation no longer fits in execution memory, Spark writes part of it to local disk and reads it back later, so the task finishes, only slower..

Fewer shuffles

The best shuffle is the one that does not run. A join with a small table can be a broadcast join, which sends the small table to every executor and leaves the big one where it is. Filtering and selecting columns before a join shrinks what gets shuffled. A table bucketed or partitioned on the join key can skip the Exchange entirely. When 2 big tables must meet, Spark shuffles both and runs a sort-merge joinSort-Merge JoinSpark's default join for 2 large tables: it shuffles both by the join key, sorts each partition, then walks the 2 sorted sides together to find matching rows.. The shuffle partition calculator works out the partitions for a given shuffle size, and our Spark UI guide walks through a real shuffle on the stage page.