Shuffle Partition Count: Why 200 Is Wrong for Your Job

Level 10: Distributed Compute & Data Operationsmedium26 minspark.sql.shuffle.partitionspartitions per core heuristicrepartition vs coalesceAQE partition coalescingCROSS JOIN with a config tableCASE classificationinteger arithmetic

Size spark.sql.shuffle.partitions with the 2-to-4-per-core heuristic, spot over- and under-partitioned stages from task metrics, and say what AQE now does for you automatically.

200 is a placeholder, not an answer

spark.sql.shuffle.partitions decides how many partitions come out the other side of every shuffle, and therefore how many tasks the next stage runs. Its default is 200, and 200 was never chosen for your data. It is wrong in both directions:

Tiny data, 200 partitions. Each task gets a few hundred kilobytes. The time to schedule a task, ship it to an executor, open a shuffle file, and report back dwarfs the time to process the rows. The job spends its life in overhead, and the tell in the metrics is a stage full of tasks that finish in tens of milliseconds.
Huge data, 200 partitions. Each task gets gigabytes. It runs out of execution memory, spills to local disk, and either crawls or dies with an executor OOM. The tell is a large spill_mb and very long task durations.

The heuristic worth memorizing

Aim for 2 to 4 partitions per total core in the cluster. Total cores means every executor's cores added up, because a core is a task slot and only one task runs per slot at a time. Two per core gives every slot a couple of waves of work, which is enough to keep everything busy and to absorb one slow task without leaving the cluster idle. Much past four and you are paying scheduling overhead for parallelism you cannot use.

On a 32-core cluster that is 64 to 128 partitions. The default of 200 is not catastrophic there, but it is a number someone else picked.

repartition and coalesce are not the same tool

Table

coalesce is cheap because it never moves rows across the network; it just glues adjacent partitions together, which is also why it can leave you with uneven ones.
operation	direction	shuffles?	when you reach for it
repartition(n)	up or down	yes, a full shuffle	you need an even redistribution, or you are partitioning by a column
coalesce(n)	down only	no, it merges neighbours	you are writing out and want fewer, bigger output files

coalesce is cheap because it never moves rows across the network; it just glues adjacent partitions together, which is also why it can leave you with uneven ones.

The trap: coalesce avoids the shuffle by merging neighbouring partitions in place, so if those neighbours were uneven, the result is uneven too. repartition costs a shuffle and gives you even partitions. Cheap and lumpy, or expensive and level.

AQE already does half of this

Adaptive Query Execution has been on by default since Spark 3.2 and Glue 4.0, and one of the things it does is coalesce post-shuffle partitions: it looks at the actual sizes after the shuffle and glues small ones together toward a target size. So the old advice, "always tune shuffle.partitions by hand", is half obsolete. An answer that never mentions AQE reads as 2019-era.

The target it aims at is spark.sql.adaptive.advisoryPartitionSizeInBytes, and its default is 64 MB. Teams and platforms raise it routinely, so when you see a bigger number in a runbook, read it as a setting somebody chose rather than as the default. The hard drill below uses a configured 128 MB target for exactly that reason.

What AQE does not do is rescue you from a badly chosen partition count on the way IN to an expensive stage, and it cannot coalesce partitions that were never small. It also only ever merges neighbours, so it can lower a stage's partition count and never raise it: if a stage already ran 200 tasks, coalescing gives you at most 200 partitions no matter how many gigabytes each one held. Knowing the heuristic still matters; you just get to say "and AQE handles the small-partition case for me".

Currency note: Spark 4.0 shipped in mid-2025 and is generally available on EMR. Its headline change is ANSI mode on by default, so divide-by-zero, overflow, and bad casts now fail fast instead of quietly returning null. That does not change partition tuning, but it does change the answer to every classic "why did my column go null" question.

Common mistake: counting executors instead of cores. Eight executors is not eight task slots; eight executors with four cores each is thirty-two, and the heuristic is per core.

Interview nuance: "your job runs 200 tasks on an 8-core cluster, what do you change" is asked almost verbatim. The full answer is three sentences: 200 is the default, not a decision; 2 to 4 per core puts you at 16 to 32 here; and AQE will coalesce the small ones at runtime anyway, so I would confirm what actually ran before hand-tuning.

On a real platform this differs. Warehouses hide this knob entirely. Snowflake sizes the compute for you by warehouse size, BigQuery allocates slots dynamically, and Redshift ties parallelism to node slices. The concept survives the migration: too few parallel units means giant units that spill, too many means overhead, and the sweet spot is a small multiple of the slots you actually have.

Sample data for this example

CREATE TABLE spark_stages (
  stage_id        INTEGER,
  name            TEXT,
  num_tasks       INTEGER,   -- one task per shuffle partition on the read side
  shuffle_read_mb INTEGER
);
INSERT INTO spark_stages (stage_id, name, num_tasks, shuffle_read_mb) VALUES
  (11, 'filter_events',       200,     60),
  (12, 'join_events_users',   200, 240000),
  (13, 'agg_daily',           200,  24000),
  (14, 'distinct_sessions',   200,    100),
  (15, 'write_curated',       200,  40000),
  (16, 'join_wide_dim',       200, 300000),
  (17, 'map_lookup',          200,    200),
  (18, 'repartition_by_user', 200, 200000);
CREATE TABLE cluster_config (
  setting TEXT,
  value   TEXT     -- every setting arrives as text, exactly as the Environment tab prints it,
                   -- so memory carries its unit suffix and byte counts do not
);
INSERT INTO cluster_config (setting, value) VALUES
  ('spark.sql.autoBroadcastJoinThreshold', '10485760'),
  ('spark.driver.memory',                  '4g'),
  ('spark.executor.memory',                '8g'),
  ('spark.executor.cores',                 '4'),
  ('spark.sql.shuffle.partitions',         '200'),
  ('spark.sql.adaptive.enabled',           'true');

Worked example (SQL)

-- Every stage's average megabytes per task, next to the partition count the job is configured with.
SELECT s.stage_id, s.name, s.num_tasks,
       ROUND(s.shuffle_read_mb * 1.0 / s.num_tasks, 2) AS mb_per_task,
       c.value AS configured_partitions
FROM spark_stages s
CROSS JOIN cluster_config c
WHERE c.setting = 'spark.sql.shuffle.partitions'
ORDER BY mb_per_task DESC;

Apply

Your turn

The task this lesson builds to.

Write a query that returns the cluster's recommended shuffle-partition range next to the value it is actually configured with, as (total_cores, low_recommendation, high_recommendation, configured), over spark_executors(executor_id, host, cores, memory_mb) and cluster_config(setting, value).

total_cores is every executor's cores added up. low_recommendation is 2 times that, high_recommendation is 4 times that. configured is the spark.sql.shuffle.partitions setting, cast to an integer. One row out. Alias the columns exactly.

3 hints and 1 automated check are waiting in the workspace.

Practice

Make it stick

A second problem on the same idea, plus 3 bonus drills.

Write a query that classifies every stage by how much data each of its tasks handled, as (stage_id, avg_mb_per_task, verdict), in stage_id order, over spark_stages(stage_id, name, num_tasks, shuffle_read_mb).

avg_mb_per_task is shuffle_read_mb divided by num_tasks, rounded to 2 decimals. verdict is 'over_partitioned' below 1 MB per task, 'under_partitioned' above 1000 MB per task, and 'ok' in between. A stage sitting exactly on either boundary is 'ok'.

2 hints and 1 automated check are waiting in the workspace.