Spark Adaptive Query Execution

Adaptive Query Execution, called AQE, lets Spark change its plan while a query runs. Normal planning happens before any data moves, so Spark must guess sizes and counts. AQE measures real numbers after each stage and corrects the remaining plan. Spark 3.2 and later versions turn AQE on by default.

The Layman View

A driver plans a trip using a map. Traffic appears on the road. A driver with a live navigation app changes the route in the middle of the trip. AQE gives Spark that live navigation.

Static Plan Versus Adaptive Plan

Static plan:
Plan once --> Stage 1 --> Stage 2 --> Stage 3 --> Result
(guesses sizes, never changes)

Adaptive plan:
Plan --> Stage 1 --> measure real sizes --> replan --> Stage 2 --> measure --> replan --> Stage 3 --> Result

Spark measures at shuffle boundaries. Each shuffle writes files, and the file sizes reveal the truth about the data.

Switching On AQE

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

Verify it with explain. The plan starts with the label AdaptiveSparkPlan when AQE is active.

Feature 1: Coalescing Shuffle Partitions

The setting spark.sql.shuffle.partitions defaults to 200. A small dataset split into 200 partitions creates 200 tiny tasks and wastes time on scheduling. AQE merges small neighbouring partitions into fewer, larger ones.

Before AQE:  [5MB][2MB][1MB][4MB][3MB][2MB] ... 200 tiny partitions

After AQE:   [11MB]      [10MB]      ...    a handful of right-sized partitions

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

Feature 2: Switching the Join Strategy

Spark estimates table sizes before the query starts. A filter can shrink a table far below the estimate. AQE sees the real size after the filter and switches a slow sort-merge join to a fast broadcast join.

Planned:  big table (estimated 5 GB) JOIN orders  --> sort-merge join
Reality:  after filter the table holds only 8 MB
AQE:      switches to broadcast join, skips the heavy shuffle

Feature 3: Splitting Skewed Partitions

One huge key can overload a single task. AQE detects a partition much larger than the median and cuts it into smaller pieces. Each piece joins with a copy of the matching data.

Without AQE skew handling:
Task 1: 10 MB   Task 2: 12 MB   Task 3: 9 MB   Task 4: 2 GB  <-- slow straggler

With AQE skew handling:
Task 1: 10 MB   Task 2: 12 MB   Task 3: 9 MB
Task 4a: 500 MB Task 4b: 500 MB Task 4c: 500 MB Task 4d: 500 MB

spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256m")

A partition counts as skewed when it exceeds both the factor times the median size and the byte threshold.

Reading the Adaptive Plan

result = big.join(small, "id")
result.collect()
result.explain()

AdaptiveSparkPlan isFinalPlan=true
+- BroadcastHashJoin ...

The value isFinalPlan=true appears after the query finishes and shows the plan that actually ran. The Spark UI SQL tab draws the same final plan.

Limits of AQE

  • AQE acts only at shuffle boundaries. A query without a shuffle gains nothing.
  • Streaming queries do not use AQE.
  • AQE reduces the need for manual tuning of shuffle partitions but does not replace good data design.
  • Extreme skew on a single key in a group-by still needs a design fix such as salting.

Practical Advice

  • Leave AQE enabled in every modern Spark version.
  • Set the advisory partition size to match your storage block size, often 128 MB.
  • Compare the Spark UI task times before and after enabling a setting to confirm the gain.
Previous lesson
Back to course
Next lesson