Spark AQE settings: what each one does and when to change it

Adaptive query execution, AQE, has been on by default since Spark 3.2.0, and most jobs should leave it alone. It runs a query 1 shuffle at a time, measures what each shuffle wrote, and re-plans the rest: it merges small partitions, switches a sort-merge join to a broadcast when one side turns out small, and splits a skewed partition of a join. 2 settings are worth changing often: parallelismFirst on a shared cluster, and the skew threshold when skew goes undetected.

This article lists every setting behind those 3 features, with its default in Apache Spark and on Databricks, and shows on real plans from Spark 4.2 what each one changes.

See what it does first

Each feature leaves a mark in the final plan. Here are all 3, on queries we ran on Spark 4.2.0 with 4 cores:

SortMergeJoin [customer_id], [customer_id], Inner
:- Sort
:  +- Exchange hashpartitioning(customer_id, 16)

To see the marks on your own query, run it and open it in the Spark UI's SQL / DataFrame tab, or call explain() on the same DataFrame after it ran. Before the run, explain() prints isFinalPlan=false and none of these marks; SQL EXPLAIN, which never runs the query, shows only the initial plan. How to read a Spark physical plan walks through one line by line.

The master switch

AQE applies to batch queries with at least 1 shuffle, broadcast or subquery. It does not apply to Structured Streaming, and it never reorders joins: that is the cost-based optimizer's job, from table statistics.

Merging small partitions

A shuffle writes spark.sql.shuffle.partitions partitions, 200 unless set. For a small shuffle that is 200 tasks with almost nothing to do. In the widget's first query, a groupBy over 10 million orders produced 201 partial sums, and AQE merged the 200 planned tasks into 1.

When to change it. parallelismFirst at true favours speed for 1 job on an empty cluster: many small tasks, every core busy. On a cluster that many jobs share, those small tasks compete for cores and each pays its start-up cost; Spark's own docs recommend false there, so merged partitions grow toward 64 MB.

spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")

On Databricks merged partitions already aim at 64 MB, close to but no larger, and never fewer than 2 per core (minPartitionNum, which Databricks says not to set). Databricks also accepts spark.sql.shuffle.partitions set to auto, which picks the number from the plan and the input size.

Merging cannot help a partition that is too big. If you set 200 partitions for a 2 TB shuffle, each is 10 GB and AQE leaves them alone; set the number high and let AQE merge down. Spark shuffle partitions gives the rule for the number.

Switching the join

Before a query runs, Spark sizes each side of a join from estimates. A filter that leaves 103 rows of 10 million looks, to the planner, like a big table, so it plans 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.. After the shuffles ran, AQE sees the real size and switches to a broadcast hash join, skipping both sorts and reading the shuffle files locally.

When to change it. Rarely. The switch comes after both sides were shuffled, so if you know a side is small, a hint is better: orders.join(F.broadcast(customers), "customer_id") plans the broadcast before anything runs, and Databricks gives the same advice. If a small table is still not broadcast, check the join type first: a left outer join cannot broadcast its left side.

Splitting skewed partitions

When 1 key holds most of the rows, 1 partition of a join holds most of the data, and 1 task runs while the rest wait. AQE splits that partition into pieces near the advisory size and joins each with a copy of the matching rows from the other side. In the widget's skew example the slowest task fell from 1,238 ms to 414 ms.

When to change it. When the Spark UI shows 1 task far slower than the median and the plan has no skew=true, the partition missed 1 of the 2 tests. The usual miss is the 256 MB floor: with many partitions, a heavy one can be 20 times the median and still smaller. Lower the threshold, keeping it above the advisory size as Spark's docs advise:

spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "128MB")

Apache Spark splits skew only inside sort-merge joins; Databricks also handles shuffled hash joins. Neither splits a groupBy, where every row of a key must meet in 1 task: salting the key is the fix there. Data 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 how to handle skewed data in PySpark cover both cases.

What to leave alone

• spark.sql.adaptive.enabled. Turning AQE off brings back 200 tiny tasks, missed broadcasts and unsplit skew.
• minPartitionNum on Databricks. Databricks says not to set it, because it overrides the minimum partition size.
• The skew factor. 5 times the median is a sound test; the size floor is the one that misses.
• Hints AQE already covers. Databricks recommends AQE's skew handling over the old skew hint.

Check it worked

After a change, run the query and compare the final plan with the initial one: AQEShuffleRead coalesced for merged partitions, AQEShuffleRead local for a switched join, AQEShuffleRead skewed with SortMergeJoin(skew=true) for a split. Then open the stage and compare the slowest task with the median. Our Spark UI guide shows both on a skewed join, before and after, and 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. explains the 3 features in short.