Adaptive Query Execution

The problem it solves

The optimizer picks join strategies and partition counts from estimates. Estimates are often wrong: a filter removes 99% of rows, or one key holds half the data. A plan fixed up front then shuffles into 200 nearly empty partitions, or sends one giant partition to a single task.

How it works

At every shuffle boundary, Spark pauses, measures the shuffle output, and re-optimizes the rest of the plan.

1. Step 1

   A stage runs and writes its shuffle

2. →

   Step 2

   Spark reads the real partition sizes

3. →

   Step 3

   The remaining plan is re-optimized

Re-optimizing can change three things:

Walk through one query

Daily revenue for one country, from a 500 GB orders table joined to customers:

SELECT o.order_date, SUM(o.amount)
FROM orders o JOIN customers c ON o.customer_id = c.id
WHERE c.country = 'NZ'
GROUP BY o.order_date

The arithmetic behind the second row: 3 GB divided by the 64 MB advisory size is about 47. Fewer, fuller tasks mean less scheduling overhead and fewer tiny output files, with no change to your code. When AQE converts a join to broadcast after the shuffle already ran, it can also read the shuffle files locally on each executor (a "local shuffle reader") instead of fetching them over the network.

Key configs

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")

SET spark.sql.adaptive.enabled = true;
SET spark.sql.adaptive.advisoryPartitionSizeInBytes = 128MB;

Spot it in a plan

A plan that AQE manages starts with AdaptiveSparkPlan. Before an action runs it shows isFinalPlan=false; after it runs, the SQL tab of the Spark UI shows the final, re-optimized plan, with nodes such as AQEShuffleRead where partitions were coalesced or split.

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [customer_id], [id], Inner
   ...

When AQE cannot help

AQE only acts at shuffle boundaries, using statistics it has already measured. That leaves real gaps:

• The first stage. Nothing has been measured yet, so the file scan and anything before the first shuffle run as planned.
• Queries with no shuffle. A filter-and-write job has no boundary to re-plan at.
• Skewed aggregations. Skew handling splits partitions in sort-merge joins. A hot key in a groupBy still lands on one task. See Data skew and salting.
• Structured Streaming. Streaming queries do not use AQE.

Common mistakes