When Spark's Static Plan Fails: Tuning Adaptive Query Execution for Skewed Joins

The straggler problem: when one task ruins your query

You submit a distributed join, watch 199 tasks finish in under a minute, and then wait twenty minutes for a single straggler. The culprit is almost always data skew: one shuffle partition contains a dominant key, so one executor processes a vastly larger share of the work while the rest of the cluster sits idle.

Adaptive Query Execution (AQE) exists to fix exactly this class of problem. Instead of trusting the physical plan generated before execution starts, Spark re-optimizes the plan at runtime using statistics collected after shuffle boundaries. The useful takeaway is that AQE can turn a skewed join straggler into balanced parallel work, but only when you understand its limits and configure it deliberately.

What AQE actually re-plans

A shuffle boundary is the point where Spark repartitions data across the cluster—typically before an aggregation or a join. Before AQE, Spark committed to a fixed number of shuffle partitions based on table statistics that were often stale or missing. AQE defers that commitment. After the first shuffle stage materializes, Spark inspects the actual partition sizes and may:

• Coalesce small partitions so that dozens of tiny tasks do not each pay fixed scheduling overhead.
• Split skewed partitions in sort-merge joins, replicating the smaller side of the join so the dominant key is processed in parallel subpartitions.
• Prune partitions dynamically in star-schema queries when a build-side filter reduces the relevant data to a tiny subset.

All three behaviors depend on runtime feedback, which means the final physical plan can differ from the one you see in a static EXPLAIN.

Worked example: a skewed customer-order join

Imagine joining a large orders table with a smaller customers table on customer_id. A handful of enterprise accounts generate most orders, so the shuffle partition containing those keys becomes the bottleneck.

Configuration

Run the following on your Spark submission gateway or cluster edge node. You need permission to submit applications with custom Spark properties. Replace the placeholders with values that match your executor memory profile.

spark-submit \
  --class com.example.SkewedJoinJob \
  --conf spark.sql.adaptive.enabled=true \
  --conf spark.sql.adaptive.coalescePartitions.enabled=true \
  --conf spark.sql.adaptive.coalescePartitions.initialPartitionNum=<num-initial-partitions> \
  --conf spark.sql.adaptive.skewJoin.enabled=true \
  --conf spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes=<bytes> \
  --conf spark.sql.autoBroadcastJoinThreshold=-1 \
  /path/to/application.jar

Setting spark.sql.autoBroadcastJoinThreshold=-1 disables broadcast hash joins for this example so the planner chooses a sort-merge join. That is a deliberate trade-off: if your build side is genuinely small, disabling broadcast may slow the query down. The setting is useful here only because skew-join optimization applies exclusively to sort-merge joins; if the planner picks a broadcast hash join, AQE will not split the straggler partition.

What Spark does at runtime

With AQE enabled, Spark still starts with the static plan. After the shuffle feeding the join completes, it measures each partition. When one partition exceeds the skew threshold and is significantly larger than the median, Spark splits it into subpartitions and replicates the smaller customers side so each subpartition can be processed independently. The final plan contains an AdaptiveSparkPlan wrapper around the re-optimized exchange.

How to verify the final plan

Do not assume the configuration took effect. Check it explicitly.

In spark-shell, PySpark, or a notebook cell, run an EXPLAIN against the target catalog. No special permissions are required beyond read access to the tables.

EXPLAIN FORMATTED
SELECT ...

Look for an AdaptiveSparkPlan node in the output. If it is absent, either AQE is disabled or the query never reaches a shuffle boundary that triggers re-planning.

Next, open the Spark UI and navigate to the SQL tab. Find the completed query and inspect the final physical plan. Confirm that the exchange feeding the join now shows a different partition count than the initial static plan, and compare per-task input sizes across the stage. The expected check is a flatter distribution of input records and a shorter maximum task duration.

Trade-offs you need to own

AQE is not free. Re-planning consumes cluster resources: Spark must materialize shuffle output, collect statistics, and re-generate the plan before downstream stages proceed. For very short queries, that overhead can outweigh the benefit.

Plan instability is the more subtle cost. Because the final plan depends on runtime data, two identical queries can produce different physical plans. This breaks regression tests that assert exact plan shapes or deterministic output partitioning, which is a common pain point in CI pipelines.

Behavior is also version-sensitive. Defaults and sub-features changed across Spark 3.x releases, so confirm every setting against your exact distribution rather than assuming a uniform baseline. What works on Spark 3.3 may behave differently on 3.5.

A practical default posture

Leave AQE enabled for general ETL workloads, but do not trust the out-of-the-box defaults blindly. Explicitly set spark.sql.adaptive.coalescePartitions.initialPartitionNum and the target size to match your executor memory profile. For joins that are chronically skewed, enable skew-join optimization and set a threshold that reflects your actual data distribution.

Before promoting a change to production, run the query with AQE enabled and disabled, then compare wall-clock time and peak execution memory for a representative dataset. If the Spark UI shows balanced task input sizes and the EXPLAIN output contains an AdaptiveSparkPlan, the runtime re-planning is doing what you expect.