Adaptive Query Execution (AQE) Tuning Guide

Datanest Digital — Spark Optimization Playbook

AQE is Spark's runtime query re-optimization engine. It observes actual data statistics during execution and adjusts the query plan on the fly. This guide covers every AQE feature, when it helps, and how to tune it.

What AQE Does

AQE re-optimizes the query plan at stage boundaries (after each shuffle). It can:

1. Coalesce shuffle partitions — Merge small post-shuffle partitions into larger ones

2. Switch join strategies — Convert sort-merge join to broadcast join at runtime

3. Handle skewed joins — Split skewed partitions and replicate the other side

4. Optimize skewed aggregations — Split skewed groups across multiple tasks

Enabling AQE

# Master switch (enabled by default on DBR 12.2+)
spark.conf.set("spark.sql.adaptive.enabled", "true")

AQE is generally safe to enable and should be on for all workloads. The only reason to disable it is for benchmarking to isolate its impact.

Feature 1: Partition Coalescing

What It Does

After a shuffle, some partitions may be tiny (a few KB). AQE merges adjacent small partitions into larger ones to reduce task overhead.

Key Configs

Tuning Strategy

# For large datasets (> 100 GB after shuffle):
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.initialPartitionNum", "4096")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "134217728")  # 128 MB
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", "4194304")  # 4 MB

# For small datasets (< 10 GB after shuffle):
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "67108864")  # 64 MB
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", "1048576")  # 1 MB

When to Increase initialPartitionNum

• Set higher than expected partition count to give AQE room to coalesce down
• Rule of thumb: data_size_bytes / advisoryPartitionSizeInBytes * 3
• Excess partitions have negligible cost — AQE will merge them

Feature 2: Dynamic Join Strategy Switching

What It Does

When AQE discovers that one side of a sort-merge join is small enough to broadcast (below the broadcast threshold), it dynamically converts the join to a broadcast hash join — eliminating the shuffle on the large side.

Key Configs

Tuning Strategy

# Set static threshold conservatively
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")  # 100 MB

# AQE will additionally convert joins at runtime when it discovers
# actual table sizes — no additional config needed beyond enabling AQE

Important Notes

• AQE's dynamic switching uses actual runtime sizes (post-filter, post-aggregation), so it catches cases the static planner misses
• If a table has stale statistics, AQE compensates automatically
• Monitor the SQL tab in Spark UI — look for "BroadcastHashJoin" replacing "SortMergeJoin"

Feature 3: Skew Join Optimization

What It Does

Detects partitions that are significantly larger than the median. Splits the skewed partition into sub-partitions and replicates the corresponding partition from the other side of the join.

Key Configs

Tuning Strategy

# Default settings work well for most cases
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")

# For highly skewed data (e.g., 100:1 skew ratio):
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "134217728")  # 128 MB

# For moderate skew (10:1 ratio), tighten the detection:
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "3")

Limitations

• Only works for sort-merge joins (not broadcast joins — those don't have the skew problem)
• Only handles skew on one side of the join at a time
• If both sides are skewed on the same key, consider salting instead (see optimization_patterns.md)

Feature 4: Skewed Aggregation Optimization

What It Does

For GROUP BY operations with skewed groups, AQE can split the aggregation into two phases: a partial aggregation on sub-partitions, then a final merge.

Key Configs

This is generally handled automatically when AQE is enabled.

AQE Configuration Template

Copy-paste this block as a starting point for your cluster init scripts or notebook headers:

# --- AQE Configuration — Datanest Digital Recommended ---

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

# Partition coalescing
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.initialPartitionNum", "4096")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "134217728")  # 128 MB
spark.conf.set("spark.sql.adaptive.coalescePartitions.minPartitionSize", "4194304")  # 4 MB

# Join optimization
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "104857600")  # 100 MB

# Skew handling
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", "268435456")  # 256 MB

Verifying AQE Is Working

In the Spark UI SQL Tab

1. Open the query's SQL plan in the Spark UI

2. Look for AdaptiveSparkPlan as the root node

3. Look for annotations:

• coalesced partitions: X -> Y (coalescing happened)
• BroadcastHashJoin where you expected SortMergeJoin (dynamic switching)
• skewed annotations on join nodes (skew handling)

Programmatically

# Check if AQE is active
print(spark.conf.get("spark.sql.adaptive.enabled"))

# After running a query, check the plan
df = spark.table("my_table").groupBy("key").count()
df.explain(mode="formatted")
# Look for "AdaptiveSparkPlan" in the output

Common Pitfalls

1. Setting shuffle.partitions too low with AQE — AQE can only coalesce (merge) partitions, not split them. Set initialPartitionNum high.

2. Disabling coalescing but keeping AQE — You lose the most impactful feature.

3. Expecting AQE to fix all skew — AQE skew handling is limited to sort-merge joins. Use salting for complex skew patterns.

4. Ignoring AQE overhead — AQE adds milliseconds of planning overhead per stage. For sub-second queries, this can be significant. Disable for very fast queries.

*AQE Tuning Guide — Datanest Digital Spark Optimization Playbook*

*https://datanest.dev*

Cluster Sizing Guide

Datanest Digital — Spark Optimization Playbook

A decision-tree approach to selecting the right Databricks cluster configuration based on your workload characteristics, data volume, and budget constraints.

Core Principles

1. Right-size, don't over-provision. Larger clusters cost more per DBU-hour and often hit diminishing returns.

2. Match instance type to workload. Compute-optimized vs. memory-optimized vs. storage-optimized matters.

3. Use autoscaling for variable workloads. Set min/max workers based on peak and trough.

4. Start small, measure, adjust. Use the benchmarking framework to validate sizing decisions.

Decision Tree: Instance Family Selection

What is your PRIMARY workload?
│
├── ETL / Data Transformation
│   ├── Shuffle-heavy (joins, groupBy, window functions)?
│   │   ├── YES → Memory-optimized (r5d, Standard_E series, n2-highmem)
│   │   └── NO  → General purpose (m5d, Standard_DS, n2-standard)
│   │
│   └── Large intermediate datasets that need local disk?
│       ├── YES → Storage-optimized (i3, Standard_L series)
│       └── NO  → General purpose
│
├── Machine Learning Training
│   ├── GPU required?
│   │   ├── YES → GPU instances (p3, Standard_NC, a2-highgpu)
│   │   └── NO  → Compute-optimized (c5, Standard_F series, c2-standard)
│   │
│   └── Large model / embeddings in memory?
│       ├── YES → Memory-optimized
│       └── NO  → Compute-optimized
│
├── SQL Analytics / BI Queries
│   ├── Concurrent users > 10?
│   │   ├── YES → SQL Warehouse (serverless recommended)
│   │   └── NO  → SQL Warehouse (classic) or Photon-enabled cluster
│   │
│   └── Queries scan > 1 TB?
│       ├── YES → Storage-optimized with Photon
│       └── NO  → General purpose with Photon
│
└── Streaming
    ├── Micro-batch latency < 1 minute?
    │   ├── YES → Fixed-size cluster (no autoscaling), memory-optimized
    │   └── NO  → Autoscaling cluster, general purpose
    │
    └── State management (aggregations, dedup)?
        ├── YES → Memory-optimized, size for state
        └── NO  → General purpose

Data Volume to Worker Count Mapping

Use these as starting points. Adjust based on benchmarking results.

ETL Workloads

SQL Analytics

Streaming

Memory Sizing Rules

Executor Memory Formula

Available memory per executor = (Instance Memory - OS Overhead) × spark.memory.fraction

Where:
  OS Overhead ≈ 10% of instance memory (min 384 MB)
  spark.memory.fraction default = 0.6

Key Ratios

Autoscaling Configuration

When to Use Autoscaling

Autoscaling Settings

Min workers = steady-state need (floor)
Max workers = peak burst need (ceiling)
Scale-down interval: 60-120 seconds (avoid thrashing)

Instance Type Quick Reference

AWS

Azure

GCP

Cost Optimization Tips

1. Use spot/preemptible instances for workers — 60-90% savings on worker compute. Keep the driver on-demand.

2. Auto-terminate idle clusters — Set auto-stop to 10-30 minutes for interactive clusters.

3. Right-size the driver — Driver rarely needs to be larger than a worker. Use the same or one size smaller.

4. Use Photon for scan-heavy workloads — Photon has higher DBU rates but faster execution, often net cheaper.

5. Schedule jobs during off-peak — Some cloud providers offer lower spot prices at night.

6. Use the cost_per_query_estimator — Quantify before committing to a cluster change.

*Cluster Sizing Guide — Datanest Digital Spark Optimization Playbook*

*https://datanest.dev*