`spark.sql.shuffle.partitions` and Adaptive Query Execution

Description

Every Exchange you saw in lessons 3, 5, and 6 mentioned the number 200 — Exchange hashpartitioning(store_id#2, 200). That number doesn't show up by chance: it's a specific setting's default value, spark.sql.shuffle.partitions, and it decides how many partitions any shuffle ends up with, regardless of whether the result has three rows or three billion. This lesson shows, with real evidence, why that fixed number is almost never right, and how Adaptive Query Execution (AQE) fixes it at runtime, based on real statistics instead of a value guessed ahead of time.

Connection to the module. This is the module's last conceptual piece: lessons 2 through 6 built the full model of partitions and shuffle; this lesson closes it out by explaining why the number 200 that showed up in every .explain() plan doesn't always survive into real execution, and what changes it.

An analogy: two hundred fixed stops, no matter how much cargo there is

Picture a logistics company that, by internal policy, always splits any delivery — whether one package or ten thousand — into exactly two hundred delivery stops. For a ten-thousand-package delivery, two hundred stops might make sense: each one gets fifty packages, a reasonable volume. But for a three-package delivery — one per Kiosko store, in this guide's analogy — two hundred stops is an absurd policy: one hundred ninety-seven trucks would leave completely empty, and the three real packages would each end up at its own stop, with no coordination benefit at all. The fixed policy doesn't know, ahead of time, how many packages there are going to be — it only finds out after the work has already started. Adaptive Query Execution is, precisely, the system that checks how many packages there really are, partway through, and adjusts the number of stops accordingly, instead of blindly following the fixed policy.

Worked example: a fixed 200, versus the number AQE genuinely decides

# shuffle_partitions_and_aqe.py
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

scale_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("franchise_id", IntegerType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
orders_at_scale_df = spark.read.csv(
    "kiosko_orders_at_scale.csv", schema=scale_schema, header=True, enforceSchema=False,
)

print(f"spark.sql.shuffle.partitions (default) = {spark.conf.get('spark.sql.shuffle.partitions')}")
print(f"spark.sql.adaptive.enabled (default)    = {spark.conf.get('spark.sql.adaptive.enabled')}")

def build_query():
    return (
        orders_at_scale_df
        .withColumn("revenue", F.col("quantity") * F.col("unit_price"))
        .groupBy("store_id")
        .agg(F.sum("revenue").alias("total_revenue"))
    )

print("\n=== AQE disabled ===")
spark.conf.set("spark.sql.adaptive.enabled", False)
q_no_aqe = build_query()
q_no_aqe.collect()
print(f"post-shuffle partitions (AQE off) = {q_no_aqe.rdd.getNumPartitions()}")

print("\n=== AQE enabled (the default since Spark 3.2) ===")
spark.conf.set("spark.sql.adaptive.enabled", True)
q_aqe = build_query()
q_aqe.collect()
print(f"post-shuffle partitions (AQE on)  = {q_aqe.rdd.getNumPartitions()}")

spark.stop()

What to expect. Running python3 shuffle_partitions_and_aqe.py, the output is exactly this (executed in this run):

spark.sql.shuffle.partitions (default) = 200
spark.sql.adaptive.enabled (default)    = true

=== AQE disabled ===
post-shuffle partitions (AQE off) = 200

=== AQE enabled (the default since Spark 3.2) ===
post-shuffle partitions (AQE on)  = 1

The contrast is dramatic: the same query, over the same ten million rows, ends up with 200 post-shuffle partitions when AQE is disabled, and with just one when it's enabled. 200 was the fixed policy's number — spark.sql.shuffle.partitions's value, with no adjustment at all. 1 is what AQE decided, in real time, after seeing this groupBy's combined result (three rows, a few thousand bytes) was so small that two hundred partitions would have been, overwhelmingly, empty or near-empty — the equivalent of two hundred trucks for three packages.

Diagram: the same plan, read before and after execution

flowchart TD
    subgraph SinAQE["AQE disabled -- fixed plan, no adjustment"]
        A1["Exchange hashpartitioning(store_id, 200)"] --> A2["200 output partitions,\nregardless of the real size"]
    end

    subgraph ConAQE["AQE enabled -- re-optimized plan with real statistics"]
        B1["Exchange hashpartitioning(store_id, 200)\n-- initial plan, same as above"] --> B2["AQEShuffleRead coalesced:\nSpark measures the shuffle\nresult's real size"]
        B2 --> B3["1 output partition\n-- coalesced because the result\nis tiny (3 rows)"]
    end

    style A2 fill:#f96,stroke:#333
    style B3 fill:#9c6,stroke:#333

Worked example, part 2: the difference, read in .explain()

print("\n=== explain without AQE ===")
spark.conf.set("spark.sql.adaptive.enabled", False)
build_query().explain()

print("\n=== explain with AQE ===")
spark.conf.set("spark.sql.adaptive.enabled", True)
q_aqe_2 = build_query()
q_aqe_2.collect()
q_aqe_2.explain()

What to expect (executed in this run):

=== explain without AQE ===
== Physical Plan ==
*(2) HashAggregate(keys=[store_id#2], functions=[sum(revenue#8)])
+- Exchange hashpartitioning(store_id#2, 200), ENSURE_REQUIREMENTS, [plan_id=18]
   +- *(1) HashAggregate(keys=[store_id#2], functions=[partial_sum(revenue#8)])
      +- *(1) Project [store_id#2, (cast(quantity#4 as double) * unit_price#5) AS revenue#8]
         +- FileScan csv [store_id#2,quantity#4,unit_price#5] Batched: false, ...

=== explain with AQE ===
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   +- *(2) HashAggregate(keys=[store_id#2], functions=[sum(revenue#25)])
      +- AQEShuffleRead coalesced
            +- Exchange hashpartitioning(store_id#2, 200), ENSURE_REQUIREMENTS, [plan_id=57]
               +- *(1) HashAggregate(keys=[store_id#2], functions=[partial_sum(revenue#25)])
                  +- *(1) Project [store_id#2, (cast(quantity#4 as double) * unit_price#5) AS revenue#25]
                     +- FileScan csv [store_id#2,quantity#4,unit_price#5] Batched: false, ...
+- == Initial Plan ==
   HashAggregate(keys=[store_id#2], functions=[sum(revenue#25)])
   +- Exchange hashpartitioning(store_id#2, 200), ENSURE_REQUIREMENTS, [plan_id=44]
      +- HashAggregate(keys=[store_id#2], functions=[partial_sum(revenue#25)])
         +- Project [store_id#2, (cast(quantity#4 as double) * unit_price#5) AS revenue#25]
            +- FileScan csv [store_id#2,quantity#4,unit_price#5] Batched: false, ...

Without AQE, the plan is a single, fixed version, with no "Initial" or "Final" label at all — what Catalyst planned before execution is exactly what got executed. With AQE, the plan has two explicit versions: an Initial Plan (identical to the one above, the best decision possible before executing anything) and a Final Plan, which adds a new node, AQEShuffleRead coalesced, wrapping the same Exchange. That node is the mark that AQE, after seeing the shuffle's real statistics (36 records, 2,736 bytes — the same numbers you measured in lesson 5 with the Spark UI's REST API), decided to read those 200 potential partitions as if they were one, because combining them all into a single read task is cheaper than coordinating 200 tasks for such a tiny result.

Going deeper: why 200 is a historical inheritance, not a magic number

Spark's official documentation describes spark.sql.shuffle.partitions with useful precision:

"Configures the number of partitions to use when shuffling data for joins or aggregations."

And on Adaptive Query Execution:

"Adaptive Query Execution (AQE) is an optimization technique in Spark SQL that makes use of the runtime statistics to choose the most efficient query execution plan, which is enabled by default since Apache Spark 3.2.0."

The value 200 has existed since a much older Spark version (1.1.0, per the official configuration table itself) — an era when typical clusters and data volumes were different from today's, and a fixed number, chosen to work reasonably well in the average case, was the only option available. A fixed number's problem is exactly the one this lesson's worked example showed: no fixed number is correct for every case. Two hundred partitions can be too few for a multi-terabyte shuffle (each partition would end up overloaded), and too many for a three-row result (almost all empty, with the overhead of coordinating tasks that have nothing to do).

Adaptive Query Execution solves this problem without you having to guess the right number by hand: Spark runs the shuffle with the initial plan (200 partitions, the configured value), but before moving on to the next stages, it measures how much each resulting partition actually weighs, and combines (coalesces) them when they're much smaller than needed. This is governed by a second setting, also officially documented:

That last value — 64 MB target per partition — is the piece that explains the 1 in your worked example: groupBy("store_id").sum("revenue")'s complete result — three rows — weighs a tiny fraction of 64 MB, so AQE merges the 200 potential partitions into one, not even coming close to the limit. Both enabling settings are on by default since Spark 3.2 — it isn't something you have to manually turn on in any code in this guide; it's the standard behavior of any modern SparkSession, which is why every .explain() in this module's earlier lessons already showed AdaptiveSparkPlan isFinalPlan=... without you mentioning it explicitly.

Common mistakes

Changing spark.sql.shuffle.partitions by hand, without considering AQE already adjusts it. What happens: someone, noticing 200 partitions "sound wrong" for their case, manually changes spark.sql.shuffle.partitions to a smaller number (say, 10), thinking that optimizes the result. Why it happens: before Adaptive Query Execution (Spark versions older than 3.2), manually tuning this value was, in fact, the only way to optimize it — so the habit of touching it by hand comes from an era when it was necessary. How to spot it: if your SparkSession runs on Spark 3.2 or newer (all of them in this guide, with Spark 4.2.0) and you adjust spark.sql.shuffle.partitions by hand before checking what AQE does by default, you're probably solving a problem the engine already solves on its own. How to fix it: before touching spark.sql.shuffle.partitions by hand, confirm with .rdd.getNumPartitions() after an action what number actually resulted, with AQE active — as this lesson's worked example did. Adjusting the base value still makes sense in specific cases (for example, if you know you're going to consistently process huge data), but not as an automatic reflex.

Confusing the Initial Plan with what actually executed. What happens: someone reads a .explain() with AQE active, sees the Initial Plan with 200 partitions, and reports that number as the query's real behavior, without checking the Final Plan. Why it happens: the Initial Plan shows up first when reading the plan's text top to bottom, in the == Final Plan == section (which is actually shown BEFORE the == Initial Plan == in the text), and it's easy to confuse the order it appears in with the order of importance. How to spot it: if your description of a query with AQE active doesn't mention AQEShuffleRead or any adjustment made after the initial plan, you probably stopped at the first read of the plan, without telling apart which section describes what Spark decided before executing and which describes what it decided after, with real data. How to fix it: when .explain() shows AdaptiveSparkPlan isFinalPlan=true with both sections (Final Plan and Initial Plan), the Final Plan is always the one describing real execution — the Initial Plan is only the starting point before the adjustment.

Assuming AQEShuffleRead coalesced means there was no shuffle at all. What happens: someone sees the AQEShuffleRead node in the final plan and concludes AQE eliminated the shuffle entirely, since the result ended up in a single partition. Why it happens: "a single partition" sounds like "there was no data reorganization between executors." How to spot it: if your explanation of this plan doesn't mention the Exchange hashpartitioning(store_id, 200) that still shows up inside the Final Plan, right below AQEShuffleRead, you're missing a piece — the shuffle really did happen, with the same reorganization by store_id as always; what AQE changed was how many read partitions got used to consume that shuffle's result, not whether the shuffle itself got skipped. How to fix it: always separate "was there an Exchange?" (yes, whenever the operation structurally requires one, as you confirmed in lesson 3) from "how many partitions did the shuffle's result end up in?" (a separate decision AQE can adjust after seeing the real size).

Exercises

Exercise 1 — Confirm the same contrast (AQE on/off) over .distinct(). Repeat this lesson's experiment, but with orders_at_scale_df.select("store_id").distinct() instead of groupBy().sum(). Predict whether you expect to see the same pattern (200 without AQE, a smaller number with AQE), and verify it.

def build_distinct():
    return orders_at_scale_df.select("store_id").distinct()

spark.conf.set("spark.sql.adaptive.enabled", False)
d_no_aqe = build_distinct()
d_no_aqe.collect()
print(f"post-shuffle partitions, distinct, AQE off = {d_no_aqe.rdd.getNumPartitions()}")

spark.conf.set("spark.sql.adaptive.enabled", True)
d_aqe = build_distinct()
d_aqe.collect()
print(f"post-shuffle partitions, distinct, AQE on  = {d_aqe.rdd.getNumPartitions()}")

post-shuffle partitions, distinct, AQE off = 200
post-shuffle partitions, distinct, AQE on  = 1

Exercise 2 — Check what happens with a groupBy over franchise_id (250,000 distinct values), instead of store_id (3 values). Before running anything, write down your own prediction: would you expect AQE to coalesce groupBy("franchise_id").sum("revenue")'s result into a single partition, just like it did with store_id, or into a larger number, given the result has far more rows? Then, run it and compare against your prediction.

def build_franchise_group():
    return (
        orders_at_scale_df
        .withColumn("revenue", F.col("quantity") * F.col("unit_price"))
        .groupBy("franchise_id")
        .agg(F.sum("revenue").alias("total_revenue"))
    )

f_aqe = build_franchise_group()
f_aqe.collect()
print(f"post-shuffle partitions, groupBy franchise_id, AQE on = {f_aqe.rdd.getNumPartitions()}")
print(f"result rows (unique franchise_id) = {f_aqe.count()}")

post-shuffle partitions, groupBy franchise_id, AQE on = 1
result rows (unique franchise_id) = 250000

Exercise 3 — Explain, without code, why this lesson never reports "AQE made the query run faster." In 2-3 sentences, and without using any number of seconds, explain what evidence this lesson does use to demonstrate AQE's effect, and why that evidence is more reliable than an execution time.

Summary and next step

This lesson closed out the module's conceptual arc: spark.sql.shuffle.partitions fixes, by default, 200 partitions for any shuffle, a value inherited from a static setting that doesn't know, ahead of time, how much the real result is going to weigh. Adaptive Query Execution — on by default since Spark 3.2 — fixes this at runtime: it measures the shuffle's real size after it happens, and coalesces the resulting partitions when they're much smaller than needed. You verified this with a real contrast: 200 partitions without AQE, just 1 with AQE active, over the same groupBy("store_id").sum("revenue") from previous lessons — and confirmed, by reading .explain(), that the Exchange itself never disappears; what changes is how many read partitions get used to consume its result.

Before moving on you should be able to: explain what spark.sql.shuffle.partitions controls and since which Spark version AQE has been on by default; tell the Initial Plan apart from the Final Plan in a .explain() with AQE; and explain why AQEShuffleRead coalesced doesn't mean the shuffle got skipped.

With this module's six content lessons complete — what a partition is, why certain operations shuffle, how kiosko_orders_at_scale got built, how to read the shuffle in .explain() and in the Spark UI, repartition() versus coalesce(), and spark.sql.shuffle.partitions versus AQE — lesson 8 pulls it all together into a single project verified end to end, over the full ten million rows.

Resources

• Apache Spark — SQL Performance Tuning, spark.sql.shuffle.partitions, spark.sql.adaptive.enabled, and spark.sql.adaptive.coalescePartitions.enabled properties table (the exact values and descriptions cited in this lesson, including the default-enabled date since Spark 3.2.0). spark.apache.org/docs/latest/sql-performance-tuning.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification: AQE compared True vs False over the same groupBy at scale, reading the post-shuffle partition count.