Adaptive Query Execution (AQE) & Handling Data Skew
Prior to Spark 3.0, the physical execution plan was locked in statically before runtime based on outdated or non-existent catalog statistics. AQE inspects actual shuffle statistics at runtime to re-optimize downstream stages.
- •Understand how Spark 3+ AQE re-optimizes physical query plans at stage boundaries.
- •Configure dynamic partition coalescing and skew join resolution.
- •Triage out-of-memory (OOM) failures caused by skewed join keys.
In-Depth Architecture & Mechanics
Key AQE Features
1. **Dynamically Coalescing Shuffle Partitions**: - Instead of manually tuning `spark.sql.shuffle.partitions` (default 200), AQE combines small adjacent partitions at stage boundaries to meet a target size (`spark.sql.adaptive.advisoryPartitionSizeInBytes`, e.g., 64MB).
2. **Dynamically Converting Sort-Merge Join to Broadcast Join**: - If after filtering a table's actual output size drops below `spark.sql.autoBroadcastJoinThreshold` (10MB default), AQE switches from an expensive Sort-Merge Join to a Broadcast Hash Join dynamically!
3. **Dynamically Handling Skew Joins**: - If a partition's size is N times larger than the median partition size, AQE splits the skewed partition into smaller sub-partitions and duplicates the corresponding keys in the matching table.
Stage 1 Execution ──> [ Shuffle Write Statistics ]
│
[ AQE Runtime Plan Re-Optimizer ]
│
┌──────────────────────┼──────────────────────┐
▼ ▼ ▼
Coalesce Small Convert to Broadcast Split Skewed
Partitions Join Partitionsspark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "67108864") # 64MB
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")