spark.sql.shuffle.partitions explained: what it does and how to set it

Almost every Spark job has a setting called spark.sql.shuffle.partitions, and almost nobody knows what number to put in it. Its default is 200. That is right for some jobs, wasteful for small ones and too small for very large ones. This guide shows what the setting really controls, using a small job you can run yourself, so you can reason about the number instead of guessing.

What you will learn

• What a shuffle partition is and which operations create them.
• Why 200 partitions on a tiny dataset leaves almost all of them empty.
• How the setting changes the number of output files.
• What Adaptive Query Execution (AQE) does with the number for you.
• A simple way to choose the value, and what this setting cannot fix.

1What the setting controls

Some operations need rows with the same key to be on the same machine: groupBy, join, distinct and window functions. Spark moves the rows across the network so that happens. This move is called a shuffle. After a shuffle, the data is split into a fixed number of pieces called shuffle partitions, and each piece is one task. The number of pieces is spark.sql.shuffle.partitions.

The test data is 10,000 rows spread over 5 cities. We total the amount per city. AQE is switched off for now, so you see the plain behaviour first.

import glob, os
from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.master("local[2]").getOrCreate()
spark.conf.set("spark.sql.adaptive.enabled", "false")

cities = ["Pune", "Delhi", "Chennai", "Mumbai", "Kochi"]
df = (spark.range(10000)
      .withColumn("city", F.element_at(F.array(*[F.lit(c) for c in cities]), (F.col("id") % 5 + 1).cast("int")))
      .withColumn("amount", F.col("id") % 100))

print("default setting:", spark.conf.get("spark.sql.shuffle.partitions"))
totals = df.groupBy("city").agg(F.sum("amount").alias("total"))
print("partitions after the groupBy:", totals.rdd.getNumPartitions())

default setting: 200
partitions after the groupBy: 200

Spark created 200 tasks to total 5 cities. How many of those partitions actually hold any rows?

filled = totals.withColumn("pid", F.spark_partition_id()).groupBy("pid").count().count()
print("partitions that hold rows:", filled)
print("empty partitions:", 200 - filled)

partitions that hold rows: 5
empty partitions: 195

5 partitions do all the work and 195 are empty. Each empty task is cheap, but not free: Spark has to schedule it, run it and record it. On a big cluster, thousands of needless tasks add up.

2Change the number

A smaller number means fewer, fuller partitions.

spark.conf.set("spark.sql.shuffle.partitions", "8")
totals8 = df.groupBy("city").agg(F.sum("amount").alias("total"))
print("partitions after the groupBy:", totals8.rdd.getNumPartitions())
filled8 = totals8.withColumn("pid", F.spark_partition_id()).groupBy("pid").count().count()
print("partitions that hold rows:", filled8)

partitions after the groupBy: 8
partitions that hold rows: 3

With 8 partitions, only 3 hold rows. Spark decides which partition a key goes to by hashing it, so two or more cities can share a partition and others get none. More partitions than keys never spreads the work any better, and even a number close to the key count can leave some partitions empty.

3Partitions become files

When you write the result, each partition that holds rows writes its own file. A tiny result can turn into several tiny files, which is the small file problem.

def part_files(path):
    return sorted(os.path.getsize(f) for f in glob.glob(path + "/part-*"))

spark.conf.set("spark.sql.shuffle.partitions", "200")
df.groupBy("city").agg(F.sum("amount").alias("total")).write.mode("overwrite").parquet("/tmp/guide_out_200")
files200 = part_files("/tmp/guide_out_200")
print("files written with 200 partitions:", len(files200))

spark.conf.set("spark.sql.shuffle.partitions", "8")
df.groupBy("city").agg(F.sum("amount").alias("total")).write.mode("overwrite").parquet("/tmp/guide_out_8")
files8 = part_files("/tmp/guide_out_8")
print("files written with 8 partitions:", len(files8))

files written with 200 partitions: 6
files written with 8 partitions: 3

5 rows of results became 6 files with the default setting. Five of them hold one city each, and the sixth is a small file with no rows, only the schema. 8 partitions gave 3 files.

4Adaptive Query Execution does this for you

Since Spark 3.2, AQE is on by default. After the shuffle it looks at the real sizes of the partitions and merges small neighbours into fewer, bigger ones. Switch it on and run the same job with the default of 200.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.shuffle.partitions", "200")
df.groupBy("city").agg(F.sum("amount").alias("total")).write.mode("overwrite").parquet("/tmp/guide_out_aqe")
print("files written with AQE on:", len(part_files("/tmp/guide_out_aqe")))

files written with AQE on: 1

1 file. AQE merged the tiny partitions after the shuffle, so the job was still planned with 200 but wrote 1 file. That is why the default of 200 hurts less than it used to. AQE only merges; it cannot split a partition that is too big unless skew handling applies, so a number that is too small is still your problem.

5How to choose a value

Start from the size of the data after the shuffle, not from the number of rows. A common rule of thumb is to aim for roughly 100 to 200 MB per partition, then check the Spark UI to see if tasks are very short (too many partitions) or spill to disk (too few). The reasoning:

These figures come from dividing the size by 100 to 200 MB. They are a starting point to measure against, not a rule that fits every job. Tasks that finish in a fraction of a second suggest too many partitions. Tasks that spill to disk or run out of memory suggest too few.

What this setting cannot fix

Where the setting applies

It applies after a shuffle caused by groupBy, join, distinct and window functions. It does not change how many partitions a file read produces, and df.repartition(3, "city") uses the 3 you give it. A repartition by column with no number uses this setting:

spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.conf.set("spark.sql.shuffle.partitions", "200")
print("repartition by column, no number:", df.repartition("city").rdd.getNumPartitions())
print("repartition with an explicit 3:  ", df.repartition(3, "city").rdd.getNumPartitions())

repartition by column, no number: 200
repartition with an explicit 3:   3

6Check yourself

• You group 10,000 rows into 5 cities with the default setting. Roughly how many tasks are empty?
• Why can an 8-partition job still leave some partitions empty when there are 5 keys?
• What did AQE change about the output files in this guide?
• One customer has 90% of the rows. Will raising the setting to 2,000 help? Why not?

Practise it: try Repartition vs Coalesce and Small Files Problem in the PySpark editor.