What should spark.sql.shuffle.partitions be?

Enough that no partition is much over 128 MB and every core gets at least 2 tasks. For 200 GB on 64 cores that is 1,600, about 128 MB each. At the default of 200, each partition would hold 1 GB, large enough to spill to disk. On Databricks, setting it to auto lets Databricks choose the number for each query.

Does adaptive query execution set the shuffle partitions for me?

Partly. It merges small neighbouring partitions toward 64 MB after the shuffle, so a number set too high costs little. It never adds partitions, so a number set too low stays too low. Set enough partitions for your largest shuffle and let adaptive query execution merge them for the small ones.

How does Spark decide a partition is skewed?

A partition is skewed when it is over 5 times the median partition size and over 256 MB. With 200 GB at 200 partitions and 20 percent on one key, the median is 819 MB, so the bar is 4 GB, and the 40.8 GB partition with the key is far over it.

How does adaptive query execution fix a skewed join?

It splits the skewed partition into pieces and joins each with a copy of the matching partition from the other table. In the example, the 40.8 GB partition becomes 51 tasks of about 819 MB. It works for sort-merge and shuffled hash joins, on the big table's side only in inner and left outer joins when the big table is on the left, and never in a broadcast join.

Why is my Spark stage stuck on the last few tasks?

Usually skew: one task reads far more data than the rest, and the stage waits for it. With 20 percent of 200 GB on one key in a group by, which Spark does not split, the stage takes about 13 times as long as with the data spread evenly. Sort the stage's tasks by duration in the Spark UI: one much longer than the rest is the sign.

How do I fix skew in a group by?

Adaptive query execution does not split skewed partitions outside joins, so deal with the key itself. If it is a null or a placeholder such as 0 or "unknown", filter those rows out before the group by and handle them apart, then add them back. More partitions do not help: the key still lands in one.

When does Spark broadcast a table instead of shuffling it?

When the smaller table is under spark.sql.autoBroadcastJoinThreshold, 10 MB by default, Spark sends it to every executor and the big table is joined where it is, with no shuffle. Databricks also switches a sort-merge join to a broadcast at runtime when one side turns out under 30 MB. Spark cannot broadcast the right table of a right outer join or either table of a full outer join.

How many shuffle partitions your Spark job needs

For 200 GB on 64 cores, set spark.sql.shuffle.partitions to 1,600, about 128 MB each. At the default of 200, each holds 1 GB.

Put in the data the shuffle moves, the cores that run it and how much of it sits on one key. See the partitions Spark cuts it into, how adaptive query execution merges the small ones and splits a skewed one, how long the stage then waits on its longest task, and what to set, for Apache Spark or Databricks.

Work out your shuffle Questions

Data the shuffle moves

200 GB

Shuffle Write of the stage in the Spark UI.

The smaller table in the join

50 GB

Spark broadcasts a table up to 10 MB instead of shuffling.

Data on the most common key

20%

Every row with the same key lands in the same partition: a null, a default value or one huge customer.

Cores in the cluster

Tasks that run at once: executors × cores each.

spark.sql.shuffle.partitions

200 by default. For this data and these cores: 1,600.

The answer

At 200 partitions, 200 GB shuffles into partitions of 819 MB, and the one the most common key lands in holds 40.8 GB. Each is over the 64 MB that adaptive query execution merges up to, so it keeps the 199 as they are and splits the skewed one into 51 tasks of 819 MB.

Partitions set: 200
Tasks after merging: 250
Rounds on your cores: 4
Longest task: 819 MB

Data each task reads

A typical task after adaptive query execution
819 MB
The most common key's partition before
40.8 GB
Each piece it is split into
819 MB

Red: over 256 MB, the size from which Spark can call a partition skewed. A partition counts as skewed when it is also over 5 times the median, here 4 GB.

What to change

Each partition holds 819 MB. Partitions that large are slow and spill to disk; 1,600 partitions make them 128 MB or less.

The settings

spark.sql.shuffle.partitions 1600
spark.sql.adaptive.enabled true
spark.sql.adaptive.coalescePartitions.parallelismFirst true
spark.sql.adaptive.skewJoin.enabled true

Pass each line to spark-submit as --conf name=value, or set it with spark.conf.set before the query runs. Adaptive query execution and skew join handling are on by default since Spark 3.2; they are written out here so the settings say so.

How the calculator works it out

A shuffle cuts the data into spark.sql.shuffle.partitions partitions, 200 unless set, by a hash of the key, so every row with the same key lands in the same partition. With the keys even, each partition holds the data divided by the partitions. A key that holds a share of the data puts all of that share in one partition, on top of its even part.

Adaptive query execution, on by default since Spark 3.2, then measures each partition and merges neighbours while their sum stays under a target. The target is 64 MB (advisoryPartitionSizeInBytes). On Apache Spark, parallelismFirst is on by default: Spark then aims for at least as many tasks as the cluster has cores, so the target drops to the data divided by the cores, but never under 1 MB. Merging never adds partitions: too few to start with stays too few.

A partition counts as skewed when it is over 5 times the median partition and over 256 MB. In a sort-merge or shuffled hash join, Spark splits it into pieces near the larger of 64 MB and the average partition that is not skewed, and joins each piece with a copy of the matching partition from the other side. It can split the left side only in inner and left outer joins, and it never splits a skewed partition in a group by or a window. The calculator puts the big table on the left. The pieces come from the map tasks' output, so there are at most as many pieces as map tasks.

The time a stage takes is estimated from the data each task reads: at least the data divided by the cores, and at least the longest task. It is set against the same data spread evenly, so a skewed key the stage cannot split shows as how many times longer it runs.

The partitions to set follow the Spark sizing calculator: no partition over 128 MB, at least 2 tasks a core as Spark's tuning guide advises, rounded up to whole rounds of your cores. A join with a table under 10 MB skips the shuffle: Spark sends that table to every executor instead, and Databricks also switches a planned join to a broadcast at runtime under 30 MB.

Questions

What should spark.sql.shuffle.partitions be?: Enough that no partition is much over 128 MB and every core gets at least 2 tasks. For 200 GB on 64 cores that is 1,600, about 128 MB each. At the default of 200, each partition would hold 1 GB, large enough to spill to disk. On Databricks, setting it to auto lets Databricks choose the number for each query.
Does adaptive query execution set the shuffle partitions for me?: Partly. It merges small neighbouring partitions toward 64 MB after the shuffle, so a number set too high costs little. It never adds partitions, so a number set too low stays too low. Set enough partitions for your largest shuffle and let adaptive query execution merge them for the small ones.
How does Spark decide a partition is skewed?: A partition is skewed when it is over 5 times the median partition size and over 256 MB. With 200 GB at 200 partitions and 20 percent on one key, the median is 819 MB, so the bar is 4 GB, and the 40.8 GB partition with the key is far over it.
How does adaptive query execution fix a skewed join?: It splits the skewed partition into pieces and joins each with a copy of the matching partition from the other table. In the example, the 40.8 GB partition becomes 51 tasks of about 819 MB. It works for sort-merge and shuffled hash joins, on the big table's side only in inner and left outer joins when the big table is on the left, and never in a broadcast join.
Why is my Spark stage stuck on the last few tasks?: Usually skew: one task reads far more data than the rest, and the stage waits for it. With 20 percent of 200 GB on one key in a group by, which Spark does not split, the stage takes about 13 times as long as with the data spread evenly. Sort the stage's tasks by duration in the Spark UI: one much longer than the rest is the sign.
How do I fix skew in a group by?: Adaptive query execution does not split skewed partitions outside joins, so deal with the key itself. If it is a null or a placeholder such as 0 or "unknown", filter those rows out before the group by and handle them apart, then add them back. More partitions do not help: the key still lands in one.
When does Spark broadcast a table instead of shuffling it?: When the smaller table is under spark.sql.autoBroadcastJoinThreshold, 10 MB by default, Spark sends it to every executor and the big table is joined where it is, with no shuffle. Databricks also switches a sort-merge join to a broadcast at runtime when one side turns out under 30 MB. Spark cannot broadcast the right table of a right outer join or either table of a full outer join.
What does spark.sql.adaptive.coalescePartitions.parallelismFirst do?: When it is true, the default on Apache Spark, Spark ignores the 64 MB target while merging and keeps at least as many partitions as there are cores, down to 1 MB each. 2 GB on 64 cores then runs as 67 tasks of 30.6 MB; with it off, as 34 tasks of 60.2 MB. Spark's guide advises turning it off on a busy cluster.

Put this calculator on your own site

Copy the code into any web page: a blog post, a course page or your team's wiki. The calculator runs in a frame and shows the same figures as this page, so when the prices change here they change on your page too. It is free and needs no account, and nothing a reader types is sent anywhere.

The code opens the calculator with the settings you have chosen above.

The short script at the end lets the frame grow and shrink with the calculator. Where a page cannot take scripts, choose a fixed height and the frame scrolls.

<iframe src="https://tecyfy.com/embed/spark-shuffle-partitions" title="Spark shuffle partition calculator" width="100%" height="2300" style="border:0;display:block" loading="lazy" allow="clipboard-write"></iframe>
<p><a href="https://tecyfy.com/data-engineering/spark-shuffle-partitions">Spark shuffle partition calculator</a> by Tecyfy</p>
<script>addEventListener("message",function(e){if(e.origin!=="https://tecyfy.com"||!e.data||e.data.type!=="tecyfy:height")return;var h=Number(e.data.height);if(!(h>0))return;document.querySelectorAll("iframe").forEach(function(f){if(f.contentWindow===e.source)f.style.height=Math.min(h,20000)+"px"})})</script>

Sources