Adaptive Query Execution for Spark & Data Eng developers

Most of us have written a Spark query that seemed reasonable on small data but ground to a halt at scale, only to discover afterward that the optimizer made a poor bet about data distribution. Adaptive Query Execution for Spark & Data Eng developers flips this model: instead of committing to a fixed plan before runtime, Spark now gathers real statistics mid-execution and adjusts on the fly. For teams managing production data pipelines, this feature can mean the difference between a job completing in minutes versus hanging indefinitely.

What is Adaptive Query Execution?

Adaptive Query Execution (AQE) is Spark's mechanism for re-optimizing query execution plans after seeing actual data volumes and distributions at runtime. In Spark 2.x, the planner built a static DAG based on metadata and statistics available at compile time. If those statistics were stale, missing, or simply wrong, the query would proceed with a suboptimal plan anyway.

AQE breaks that pattern by instrumenting the query to collect data shape information as intermediate stages complete. When a shuffle finishes, Spark inspects the actual number of rows, partition sizes, and data skew. It then decides: should we repartition differently? Should we switch from a broadcast join to a sort-merge join? Should we collapse multiple stages? These decisions happen at runtime, guided by real numbers rather than estimates.

Starting with Spark 3.0, AQE is enabled by default. For most teams, this means better query performance with zero configuration changes—though understanding what's happening under the hood helps when things don't go as expected.

Enabled by Default in Spark 3+

If you're running Spark 3.0 or later, you already have AQE active. You can verify this in your session:

spark.conf.get("spark.sql.adaptive.enabled")
# Returns: true (or False if explicitly disabled)

The configuration lives in spark.sql.adaptive.enabled. Most organizations keep it on because the overhead is modest—typically a few percent of execution time—while the upside can be substantial. However, we've seen teams disable it in specific scenarios: ultra-low-latency dashboards where plan stability matters more than optimality, or when dealing with buggy UDFs that behave differently in different Spark versions.

A practical approach: enable it globally in your Spark session defaults, but document your exceptions. We maintain a config matrix for different workload types—batch ETL, interactive queries, ML feature engineering—and AQE is on for all of them in production.

When you enable AQE, several sub-features activate in concert:

• Coalescing shuffle partitions—reducing partition count when stages produce small outputs
• Join strategy switching—swapping between broadcast and sort-merge joins based on actual sizes
• Skew handling—detecting imbalanced partitions and splitting them intelligently
• Cardinality estimation refinement—using observed row counts to improve downstream cost models

Each works independently, so you can tune them separately if needed.

Coalesces Shuffle Partitions

One of the most impactful AQE behaviors is shuffle partition coalescing. By default, Spark creates a number of shuffle partitions determined by spark.sql.shuffle.partitions (default 200). This is a one-size-fits-all setting: whether your intermediate result is 1 KB or 1 TB, you still get 200 partitions.

With AQE, Spark monitors what actually lands in each partition. If a shuffle produces 2 GB of data spread across 200 partitions, that's fine—each partition is ~10 MB, a reasonable size. But if it produces only 50 MB, you now have 200 partitions of 250 KB each. Reading and processing so many tiny partitions wastes time on overhead. AQE detects this and coalesces adjacent partitions into fewer, larger ones.

Consider a real scenario: joining a large transactional fact table with a small lookup table, then aggregating:

facts = spark.read.parquet("s3://data/facts/")  # 100 GB
lookups = spark.read.parquet("s3://data/lookups/")  # 50 MB

joined = facts.join(lookups, "id")
result = joined.groupBy("category").agg(sum("amount"))
result.write.parquet("s3://output/")

Without AQE, the join shuffle uses 200 partitions. The join itself is fine. But the subsequent groupBy also shuffles with 200 partitions. If the category distribution is skewed and few groups dominate, many partitions end up tiny. AQE coalesces these into 30 or 40 partitions, cutting I/O and CPU time measurably.

You can control coalescing thresholds:

spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", True)
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)
# Target size for coalesced partitions (bytes)
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", 64 * 1024 * 1024)  # 64 MB

The advisoryPartitionSizeInBytes setting is a hint. Spark tries to coalesce partitions to roughly this size, though it won't merge partitions from different executors if that creates data locality problems.

Switches Join Strategies

Join strategy selection is one of Spark's most important optimization decisions. Broadcast joins (sending one table to all workers) are fast for small tables. Sort-merge joins (shuffling both tables and merging sorted chunks) are necessary when both sides are large. The problem: at plan time, the planner estimates sizes. Those estimates are often wrong.

Classic failure mode: the planner thinks a filtered table will be 10 MB (broadcast eligible), but the actual filter result is 500 MB. The broadcast fails or degrades to disk spilling. With AQE, this is caught mid-flight.

Here's a typical case:

customers = spark.read.parquet("s3://data/customers/")  # 2 GB
recent_orders = spark.read.parquet("s3://data/orders/")  # 10 GB

# Filter orders to recent data
recent = recent_orders.filter(col("date") >= "2025-01-01")

# Join with customers
result = customers.join(recent, "customer_id")

The planner estimates recent as ~5 GB (50% of all orders). It decides on a sort-merge join. But in reality, recent is only 200 MB. AQE detects this: after the filter stage completes, it sees 200 MB of actual data. It recalculates and switches to a broadcast join, sending that 200 MB to all executors instead of shuffling both 2 GB tables.

The configuration:

spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)
# Broadcast eligible if actual size is under this threshold
spark.conf.set("spark.sql.adaptive.maxShuffledHashJoinLocalMapperNumber", 10000000)

AQE compares actual output size against spark.sql.autoBroadcastJoinThreshold. If the threshold is exceeded initially but the actual size is below it (or vice versa), AQE adjusts the plan.

Handles Skewed Joins

Data skew is a silent killer in distributed systems. A perfectly balanced join can become a bottleneck when one partition has 10x more data than others. One slow executor stalls the entire query.

AQE has a dedicated skew-handling mode. After a shuffle completes, it calculates the median partition size. Partitions significantly larger than the median are flagged as skewed. Spark then splits these partitions and replicates the corresponding partitions from the other table, effectively creating mini-joins to balance the load.

Real example: joining orders by user ID, where a few power users have thousands of orders:

orders = spark.read.parquet("s3://data/orders/")
users = spark.read.parquet("s3://data/users/")

# This join will skew if user IDs are not uniformly distributed
result = orders.join(users, "user_id")

Without skew handling, the partition containing power-user orders will take 10x longer than others. The job is bottlenecked. With AQE skew handling enabled:

spark.conf.set("spark.sql.adaptive.skewJoin.enabled", True)
spark.conf.set("spark.sql.adaptive.skewJoin.skewFactor", 5.0)  # 5x median = skewed

Spark detects that one partition is ~5x the median size. It splits that partition into smaller chunks and replicates the matching chunks from the users table. The join parallelizes better, and execution time improves dramatically—sometimes 50% or more.

The tradeoff: extra shuffling for the replication. But on skewed data, the parallelism gain outweighs this cost. We've seen jobs that hung indefinitely become routine with this enabled.

Common Mistakes

Disabling AQE globally without cause. Some teams disable AQE because they read "it can cause issues" online. In practice, most issues were fixed years ago. Disabling it means leaving performance on the table.

Setting advisoryPartitionSizeInBytes too small. If you set it to 10 MB, Spark coalesces aggressively, creating fewer but larger partitions. This can hurt parallelism on multi-terabyte datasets. A safer default is 64 MB; adjust only if you have concrete evidence.

Ignoring the Spark UI.** The Spark UI shows which AQE optimizations actually fired. If coalescing is disabled but you see "initial partitions: 200, final partitions: 50," that's coalescing at work. Check the UI to validate your assumptions.

Forgetting that AQE adds latency to the plan. For streaming jobs with micro-batch intervals under a second, AQE's overhead (re-planning) can matter. We typically disable it for true low-latency streaming, though Structured Streaming batch mode can benefit.

Not refreshing statistics. AQE works best when base table statistics are reasonably fresh. If you never run ANALYZE TABLE, the initial cost estimates are poor, limiting AQE's room to improve. Run analysis periodically on large tables.

Key Takeaways

• AQE is enabled by default in Spark 3+; it re-optimizes plans at runtime based on actual data shapes
• Shuffle partition coalescing reduces overhead by merging small partitions into larger ones
• Join strategy switching detects when broadcast is more efficient than sort-merge and adapts accordingly
• Skew handling splits imbalanced partitions and replicates corresponding data, improving parallelism on skewed keys
• The performance gains typically outweigh the re-planning overhead; disable only when you have specific reasons
• Monitor the Spark UI to see which AQE features are triggering and validate they're helping your queries
• Tune advisory partition size and skew factor based on your data and cluster characteristics, not defaults alone