How do I enable and configure AQE?

Summary

• Adaptive Query Execution (AQE) has been on by default since Apache Spark 3.2.0. spark.sql.adaptive.enabled defaults to true and acts as the umbrella switch for every AQE feature.
• Configuring AQE mostly means tuning its sub-features: post-shuffle partition coalescing, skew-join splitting and runtime join conversion. Each has its own spark.sql.adaptive.* key.
• The setting that is easy to miss is spark.sql.adaptive.coalescePartitions.parallelismFirst. Its default of true makes Spark not respect spark.sql.adaptive.advisoryPartitionSizeInBytes while coalescing, sizing partitions from the cluster's default parallelism instead, and the Spark docs recommend setting it to false on a busy cluster.
• AQE keys are runtime SQL configurations, so you can set them per session with spark.conf.set or SET, or for the whole application with --conf or spark-defaults.conf.
• Confirm AQE did something by reading the final plan: AdaptiveSparkPlan isFinalPlan=true, an AQEShuffleRead node marked coalesced, or a SortMergeJoin(skew=true).

How do I enable and configure AQE?

You probably don't need to enable it. Adaptive Query Execution is on by default in Apache Spark 3.2.0 and later, controlled by spark.sql.adaptive.enabled. The real work is configuring its sub-features so they fit your workload, then checking the final plan to confirm they fired. Start with two changes: give coalescing a large enough starting partition count, and turn off parallelismFirst on shared clusters so AQE aims for a sensible partition size instead of maximum parallelism.

What AQE does at runtime

AQE re-optimizes a query plan in the middle of execution using statistics gathered when each query stage finishes. It applies to queries that contain at least one exchange (a shuffle or broadcast) or a subquery. It can merge small post-shuffle partitions, split skewed partitions in a sort-merge join, and switch a sort-merge join to a broadcast hash join or a shuffled hash join once the real sizes are known.

Setting	Default (Spark 4.2)	What it controls
spark.sql.adaptive.enabled	true	Umbrella switch for all of AQE
spark.sql.adaptive.coalescePartitions.enabled	true	Merges contiguous small shuffle partitions
spark.sql.adaptive.coalescePartitions.parallelismFirst	true	When true, doesn't respect the advisory size; sizes partitions from the cluster's default parallelism, usually smaller than the advisory size, with minPartitionSize (1MB) as the floor
spark.sql.adaptive.advisoryPartitionSizeInBytes	64 MB	Target partition size for coalescing and skew splitting
spark.sql.adaptive.coalescePartitions.initialPartitionNum	none, so spark.sql.shuffle.partitions	Partition count before coalescing
spark.sql.adaptive.autoBroadcastJoinThreshold	same as spark.sql.autoBroadcastJoinThreshold	Runtime size limit for converting to a broadcast join
spark.sql.adaptive.skewJoin.enabled	true	Handles skew in sort-merge joins by breaking oversized partitions into smaller tasks
spark.sql.adaptive.optimizer.excludedRules	none	Comma-separated AQE rules to disable

Set the values

AQE keys are runtime SQL configurations: per-session and mutable. Set them from code between queries:

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

# Respect the advisory size when coalescing (recommended on busy clusters)
spark.conf.set("spark.sql.adaptive.coalescePartitions.parallelismFirst", "false")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")  # example target

# Start high and let AQE merge down
spark.conf.set("spark.sql.adaptive.coalescePartitions.initialPartitionNum", "2000")  # example

In SQL, SET spark.sql.adaptive.coalescePartitions.parallelismFirst = false; does the same, and RESET returns keys to their initial values. To apply settings to a whole application, pass them at submit time or put them in conf/spark-defaults.conf:

spark-submit \
  --conf spark.sql.adaptive.coalescePartitions.parallelismFirst=false \
  --conf spark.sql.adaptive.advisoryPartitionSizeInBytes=128MB \
  my_job.py

The values above are illustrations, not recommendations. Size the advisory target to your data and executors. Coalescing can only merge partitions, so the starting count has to be large enough; the Spark tuning guide says AQE picks the right shuffle partition number at runtime once you set a large enough initial number through initialPartitionNum.
To switch off one feature and keep the rest, set its own flag to false, for example spark.sql.adaptive.skewJoin.enabled, or list the rule in spark.sql.adaptive.optimizer.excludedRules.

Check that AQE changed the plan

Run the query, then open its entry in the SQL tab of the Spark UI and expand Details, or call explain() on the executed DataFrame. Look for:

• AdaptiveSparkPlan isFinalPlan=true at the root. It reads false until execution completes, and SQL EXPLAIN never runs the query, so it only shows the initial plan.
• AQEShuffleRead coalesced above an exchange, meaning partitions were merged.
• SortMergeJoin(skew=true), meaning skewed partitions were split.
• A BroadcastHashJoin in the final plan where the initial plan had a SortMergeJoin.
• Statistics(..., isRuntime=true), the runtime statistics AQE used.

A runtime switch to broadcast is useful, but the Spark docs note it isn't as efficient as planning a broadcast hash join in the first place. For a table you know is small, a BROADCAST hint or accurate table statistics still help.

How Databricks approaches AQE

AQE is enabled by default on Databricks. Setting spark.sql.shuffle.partitions to auto turns on auto-optimized shuffle, which picks the partition number from the query plan and the input data size. On serverless compute, most Spark properties can't be set; spark.sql.shuffle.partitions is one of the few that can, and its serverless default is auto. Databricks documents its own runtime broadcast threshold, spark.databricks.adaptive.autoBroadcastJoinThreshold (default 30MB). See Adaptive query execution.

FAQs

Is AQE enabled by default in Spark?

Yes. spark.sql.adaptive.enabled has defaulted to true since Apache Spark 3.2.0, and in current releases coalescing, the local shuffle reader and skew-join handling also default to true.

Does AQE make spark.sql.shuffle.partitions irrelevant?

No. Unless you set spark.sql.adaptive.coalescePartitions.initialPartitionNum, Spark uses spark.sql.shuffle.partitions as the starting count that AQE coalesces down from.

Why didn't AQE change my plan?

A query without an exchange or subquery isn't adaptive at all, and an adaptive query is re-optimized only when runtime statistics justify a different plan. Check the final plan after execution, not the output of SQL EXPLAIN.