What is a shuffle in Spark, and what is data skew?

A shuffle is the step where Spark moves rows between machines so that every row with the same key lands in the same place. A join on customer_id, a GROUP BY country and a repartition all need one: before Spark can add up a country's sales, every sale for that country has to be on one machine. Data skew is what happens when the keys are uneven, so that one key has far more rows than the others. The machine that gets that key does most of the work while the rest wait for it.

What a shuffle costs

Spark splits a table into partitions and works on each in a separate task. Reading, filtering and adding a column happen inside each partition, with no data moving. A shuffle is different: each task writes its rows out sorted by where they need to go, and the tasks of the next stage fetch their share from every machine. That is disk writes, conversion to bytes and network traffic, which is why the Spark guide calls it costly, and why the shuffles are the first place to look when a job is slow.

The number of partitions after a shuffle is spark.sql.shuffle.partitions, 200 unless set. A small job then runs as many tiny tasks. Adaptive query execution, on by default since Spark 3.2, measures the shuffle as it happens and merges small partitions, toward 64 MB each on Databricks, so the setting matters less than it did.

What skew looks like

Picture an orders table where a quarter of the rows have no customer, stored with customer_id null or 0. A join on customer_id sends all of them to one task. In the Spark UI the stage reaches 197 of 200 tasks in a minute and then sits on the last few for half an hour. Databricks gives that as the sign: sort the stage's tasks by duration, and if one took far longer than the rest, the data is skewed. The same thing happens with one huge customer, a default value such as "unknown", or a single busy day when the key is a date.

What fixes it

• Let adaptive query execution split it. With it on, Spark treats a shuffle partition as skewed when it is over 5 times the median and over 256 MB, and splits it into tasks of roughly even size. A plan that did this shows isSkew on its sort-merge join. On Databricks this is the default, and skew hints are no longer needed.
• Avoid the shuffle. A join with a small table can send the whole small table to every machine instead. Spark does this under 10 MB; Databricks switches to it at runtime under 30 MB. Joins of a large fact table to small dimension tables, as in a star schemaStar SchemaA way to arrange warehouse tables for reporting: a fact table of events and their numbers, joined by keys to dimension tables that describe them, such as customer, product and date., usually work this way.
• Deal with the heavy key itself. A join condition on a null key is never true, so rows with a null key never match in an inner join; filter them out before it. A key such as "unknown" can often be handled apart and added back.
• Check the layout. A table partitionedData Partitioning and ClusteringTwo ways to lay out a large table so a query reads only the files it needs: partitioning splits it by a column such as the date, clustering keeps similar values in the same files. by a column with very uneven values carries its skew into every job that reads it.

Why it matters for AI

Pipelines that build training sets, features and retrieval tables are mostly joins and group-bys, so they shuffle. A job that slows from 10 minutes to 2 hours as one customer grows is usually skew, and a bigger cluster does not help, because the work is still on one task. The Spark sizing calculator sizes executors for an even job; skew is the reason to check the task durations before adding machines. The shuffle partition calculator shows what one heavy key does to a stage, and whether Spark can split it.