# Adaptive query execution in Spark: broadcast joins at run time and skew joins

Adaptive query execution re-plans a Spark query between stages, from the sizes the finished stages actually wrote. It does 3 things on its own: it merges small shuffle partitions, it switches a sort-merge join to a broadcast join when 1 side turns out to be small, and it splits a skewed partition of a sort-merge join into pieces, so 1 key with most of the rows does not leave every other core waiting. explain() after an action shows the final plan beside the initial one.

- Page: https://tecyfy.com/data-engineering/course/how-spark-runs-a-job/adaptive-query-execution
- Course: Data engineering course (https://tecyfy.com/data-engineering/course), Module 3: How Spark runs a job, Adaptive query execution
- Parts of this module: Jobs, stages and tasks (https://tecyfy.com/data-engineering/course/how-spark-runs-a-job); Shuffles and partitions (https://tecyfy.com/data-engineering/course/how-spark-runs-a-job/shuffles-and-partitions); Adaptive query execution (https://tecyfy.com/data-engineering/course/how-spark-runs-a-job/adaptive-query-execution); Build it: a job report (https://tecyfy.com/data-engineering/course/how-spark-runs-a-job/build-it)

At the end of this part you can:

- Say what a broadcast join is and when Spark chooses it
- Read an adaptive plan and find what changed at run time
- Spot a skewed join in the Spark UI and say when Spark splits it

Spark plans a query before it reads a row, from estimates: how big a table is on disk, how many rows a filter keeps. Estimates are often wrong. Adaptive query execution, on by default, lets Spark change its plan between stages, once the stages that have run have reported what they wrote. [Part 2](/data-engineering/course/how-spark-runs-a-job/shuffles-and-partitions) showed the first thing it changes, merging small partitions. This part shows the other 2, on the shop's orders.

## Broadcast joins

A join needs each order next to its customer. The general way is a sort-merge join: shuffle both tables by `customer_id`, so each customer's rows from both sides land in the same partition, sort them and walk through them together. That is 2 shuffles.

If 1 side is small, there is a cheaper way. Spark sends a full copy of the small table to every task, and each task joins its own part of the large table where it is, with no shuffle of the large table at all. That is a broadcast join, and Spark plans one when it expects a side to be under `spark.sql.autoBroadcastJoinThreshold`, 10 MB by default. Part 1's join of orders to customers was planned that way, `BroadcastHashJoin` in its plan, because the customers file is 200 KB. Turn broadcasting off with `-1` and the same join shuffles 17 MB of orders, as the demo in part 1 shows.

## A join that changes while it runs

Sometimes Spark cannot know a side is small until it has computed it. Take the orders of the shop's biggest spenders, the customers whose orders add up to more than 2,000. Save this as `aqe.py`:

```python
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.master("local[4]").appName("aqe").getOrCreate()
orders = spark.read.parquet("lake/orders")
customers = spark.read.parquet("lake/customers")

revenue = orders.groupBy("customer_id").agg(F.sum("amount").alias("revenue"))
top = revenue.filter("revenue > 2000")
big_spenders = orders.join(top, "customer_id").agg(
    F.count("*").alias("orders"), F.sum("amount").alias("amount")
)
print(big_spenders.collect())
big_spenders.explain()
```

```text
[Row(orders=68160, amount=Decimal('3445280.00'))]
```

Before it runs, Spark can only guess the size of `top` from the files under it, 22 MB, over the threshold, so the initial plan is a sort-merge join. Then the aggregate runs, its shuffle reports what it wrote, and only 1,704 customers pass the filter. `explain()` after an action prints the final plan above the initial one, trimmed here:

```text
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   ...
   +- *(4) BroadcastHashJoin [customer_id#1L], [customer_id#12L], Inner, BuildRight, false, false
      :- AQEShuffleRead local
      :  +- ShuffleQueryStage 0
      +- BroadcastQueryStage 2
         +- ...
            +- *(3) HashAggregate(keys=[customer_id#12L], functions=[sum(amount#13)])
               +- AQEShuffleRead coalesced
+- == Initial Plan ==
   ...
   +- SortMergeJoin [customer_id#1L], [customer_id#12L], Inner
```

Each `QueryStage` is a stage that ran before Spark re-planned. The orders had already been shuffled by then, but `AQEShuffleRead local` means each task reads the shuffle files on its own machine rather than fetching them across the cluster, and the sort is gone. The filtered customers became a broadcast. On a single computer the saving is small; on a cluster, it saves a network transfer of every order.

On Databricks the adaptive threshold has its own setting, `spark.databricks.adaptive.autoBroadcastJoinThreshold`, 30 MB by default. Its documentation calls the merge step `CustomShuffleReader` and shows skew as `isSkew`; Spark 4.2 prints `AQEShuffleRead` and `skew=true`.

## A skewed join

Now a problem adaptive query execution can fix but not prevent. Suppose 3 in 4 of the shop's orders are from guests who check out without an account, stored with `customer_id` 0. Add this to `aqe.py`:

```python
guests = orders.withColumn(
    "customer_id", F.when(F.col("order_id") % 4 != 0, F.lit(0)).otherwise(F.col("customer_id"))
)
guests.write.mode("overwrite").parquet("lake/orders_with_guests")
orders_g = spark.read.parquet("lake/orders_with_guests")
orders_g.groupBy("customer_id").count().orderBy(F.desc("count")).show(3)
```

```text
+-----------+-------+
|customer_id|  count|
+-----------+-------+
|          0|1500040|
|      30928|     40|
|      39104|     40|
+-----------+-------+
```

1 key holds 1.5 million rows and every other key 40. In a sort-merge join, every row of a key goes to the same partition, so 1 task gets 3 quarters of the work. That is [Shuffle and Data Skew](https://tecyfy.com/concepts/data-skew). With customers small enough to broadcast, the guest rows never move and nothing goes wrong, so to see the problem, turn broadcasting off, as if the customer table were large:

```python
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "1m")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "1m")

for skew in ["false", "true"]:
    spark.conf.set("spark.sql.adaptive.skewJoin.enabled", skew)
    by_country = (
        orders_g.join(customers, "customer_id")
        .groupBy("country_id")
        .agg(F.sum("amount").alias("revenue"))
    )
    print(skew, sum(row.revenue for row in by_country.collect()))
by_country.explain()
```

```text
false 89920000.00
true 89920000.00
```

The 2 sizes are scaled down because this join shuffles only 12 MB of orders, which is tiny for Spark. A partition counts as skewed when it is larger than 5 times the median and larger than `skewedPartitionThresholdInBytes`, 256 MB by default, and Spark aims for pieces of the advisory size, 64 MB. At 1 MB, this data behaves as a large table would with the defaults.

The totals are the same, so the difference is only in how the work ran. Open the join's stage in the Stages tab, and the summary table shows it: with the fix off, the slowest task reads 8 times what the median reads.

(An interactive demo goes here on the page: https://tecyfy.com/data-engineering/course/how-spark-runs-a-job/adaptive-query-execution)

With the fix on, the final plan says so:

```text
+- *(5) SortMergeJoin(skew=true) [customer_id#53L], [customer_id#3L], Inner
   :- *(3) Sort [customer_id#53L ASC NULLS FIRST], false, 0
   :  +- AQEShuffleRead coalesced and skewed
   +- *(4) Sort [customer_id#3L ASC NULLS FIRST], false, 0
      +- AQEShuffleRead coalesced
```

The guest partition was cut into 4 pieces, and each piece was joined with its own copy of the guest's customer row.

## What it does not fix

Adaptive query execution works only with what it can measure, and only between stages:

- It merges small partitions, but does not split large ones outside a skewed join, so a shuffle with far too few partitions stays slow.
- It splits skew only in sort-merge and shuffled hash joins, and only past both limits. A slow task that is slow for another reason, such as an expensive function, looks the same in the UI but has a normal shuffle read.
- It cannot remove the cause. Here the cause is a placeholder key, and a better fix is to keep the guest orders out of the join, or give each guest order a key of its own.

On Databricks serverless, adaptive query execution is on and you cannot change its settings, since serverless lets you set only a few properties. The query profile shows the joins that ran.

[Part 4](/data-engineering/course/how-spark-runs-a-job/build-it) turns what the Spark UI showed into a script that reports every stage of a job, and flags the problems from this module.

## Check yourself

1. Your join's stage in the Spark UI has a median task time of 2 seconds and a slowest task of 9 minutes. Adaptive query execution is on, but the plan shows no skew=true. Which is the likely reason?

   A. The skewed partition is below spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes, 256 MB by default, or not 5 times the median, so Spark does not count it as skewed
   B. Adaptive query execution only works on Databricks
   C. Skew joins are only split when the join is a broadcast join
   D. The slowest task is reading from disk, which Spark cannot split

### Answers

1. A. Spark splits a partition of a sort-merge join only when it is larger than both 5 times the median partition and 256 MB. A partition of 200 MB that holds a slow key, or a slow task for another reason such as an expensive function on 1 key, is not split. Look at the shuffle read of the slowest task in the stage's summary first; if it is the size, lower the threshold for that job, and if it is not, the cause is elsewhere.

## Questions

### What is a broadcast join in Spark?

A broadcast join sends a full copy of the smaller table to every task, so each task joins its part of the large table with no shuffle of the large table at all. Spark chooses it when it expects 1 side to be under spark.sql.autoBroadcastJoinThreshold, 10 MB by default, and adaptive query execution can switch to it during the run when a side turns out to be small. You can ask for it with F.broadcast(df).

### How does Spark handle skewed joins?

With adaptive query execution on, Spark checks the partition sizes after the shuffle of a sort-merge join. A partition larger than 5 times the median and larger than 256 MB is split into smaller pieces, each joined with a copy of the matching rows from the other side, so its work is spread over several tasks. The plan then shows SortMergeJoin(skew=true).

### How do I see the final plan of an adaptive query?

Run an action on the DataFrame, then call explain() on it. The plan then says isFinalPlan=true and prints the final plan, with the query stages that ran, above the initial plan Spark made before it started. SQL EXPLAIN does not run the query, so it shows only the initial plan.

## Sources

- [Performance tuning, adaptive query execution](https://spark.apache.org/docs/latest/sql-performance-tuning.html)
- [Adaptive query execution](https://docs.databricks.com/aws/en/optimizations/aqe)
- [Set Spark configuration properties on Databricks](https://docs.databricks.com/aws/en/spark/conf)
- [pyspark.sql.functions.broadcast](https://spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.functions.broadcast.html)