Adaptive query execution in Apache Spark - Step-by-Step Execution

Concept Flow - Adaptive query execution

Start Query

↓

Initial Plan Generation

↓

Execute Stage 1

↓

Collect Runtime Stats

↓

Analyze Stats & Adjust Plan

↓

Execute Adjusted Plan

↓

Return Final Result

The query starts with an initial plan, runs part of it, collects stats, adjusts the plan based on real data, then finishes execution.

Execution Sample

Apache Spark

df = spark.read.csv('data.csv')
result = df.filter('age > 30').join(other_df, 'id')
result = result.repartition(10)
result.collect()

This code reads data, filters rows, joins with another dataset, repartitions, and collects results, triggering adaptive query execution.

Execution Table

Step	Action	Runtime Stats Collected	Plan Adjustment	Output
1	Initial plan created with default join strategy	None	None	Plan ready
2	Execute filter and partial join	Data size smaller than expected	Switch join to broadcast join	Partial results ready
3	Repartition data to 10 partitions	Partitions balanced	No change	Data repartitioned
4	Execute adjusted join with broadcast	Join faster	No change	Final results collected
5	Return final result to user	N/A	N/A	DataFrame with joined data

💡 Final result collected after adaptive plan adjustments based on runtime stats.

Variable Tracker

Variable	Start	After Step 2	After Step 3	After Step 4	Final
df	Raw CSV data	Raw CSV data	Raw CSV data	Raw CSV data	Raw CSV data
result	Not defined	Partial join result	Partial join result	Final join result	Final join result

Key Moments - 3 Insights

Why does the join strategy change during execution?

Because runtime stats showed the data was smaller than expected, so switching to a broadcast join is faster. See execution_table step 2.

What triggers repartitioning of data?

Repartitioning happens to balance data across partitions for better parallelism, as shown in execution_table step 3.

Does adaptive query execution always change the plan?

No, it only changes the plan if runtime stats suggest a better strategy. Sometimes no change is needed, as in steps 3 and 4.

Visual Quiz - 3 Questions

Look at the execution table, at which step is the join strategy changed?

A Step 1

B Step 2

C Step 3

D Step 4

💡 Hint

Check the 'Plan Adjustment' column in execution_table rows.

According to variable_tracker, what is the state of 'result' after step 3?

A Raw CSV data

B Filtered data (age > 30)

C Partial join result

D Final join result

💡 Hint

Look at the 'result' row under 'After Step 3' in variable_tracker.

If the data size was larger than expected, what would likely happen in the adaptive execution?

A Switch to shuffle join

B Switch to broadcast join

C Keep initial join strategy

D Skip join operation

💡 Hint

Refer to execution_table step 2 where data size affects join strategy.

Concept Snapshot

Adaptive Query Execution (AQE) in Spark:
- Starts with an initial query plan
- Runs part of the query and collects runtime stats
- Adjusts the plan based on actual data (e.g., join type, partitions)
- Improves performance by adapting to real data
- Final results returned after optimized execution

Full Transcript

Adaptive query execution in Apache Spark begins by creating an initial plan for the query. The query runs partially, and Spark collects runtime statistics such as data size and partition distribution. Based on these stats, Spark may adjust the query plan, for example by switching join strategies or repartitioning data to improve performance. The adjusted plan is then executed to completion, and the final result is returned. This process helps Spark optimize queries dynamically, adapting to actual data characteristics rather than relying solely on static estimates.

Practice

1.

What is the main purpose of adaptive query execution in Apache Spark?

easy

A. To improve query performance by adjusting the execution plan during runtime

B. To statically compile queries before execution

C. To disable query optimization for faster execution

D. To manually rewrite queries for better speed

Solution

1. Step 1: Understand adaptive query execution concept

   Adaptive query execution allows Spark to change the query plan dynamically while the query runs.

2. Step 2: Identify the main benefit

   This dynamic adjustment improves performance by optimizing joins and data shuffles based on actual data size.

3. Final Answer:

   To improve query performance by adjusting the execution plan during runtime -> Option A

4. Quick Check:

   Adaptive query execution = dynamic plan adjustment [OK]

Hint: Think: adaptive means changing during execution [OK]

Common Mistakes:

• Confusing adaptive execution with static query compilation
• Believing it disables optimization
• Thinking it requires manual query rewriting

2.

Which Spark configuration enables adaptive query execution?

?

easy

A. spark.sql.execution.adaptive.enabled = false

B. spark.execution.adaptive = true

C. spark.adaptive.query.execution = true

D. spark.sql.adaptive.enabled = true

Solution

1. Step 1: Recall the exact configuration key

   The correct key to enable adaptive query execution is spark.sql.adaptive.enabled.

2. Step 2: Confirm the value to enable

   Setting it to true activates adaptive query execution.

3. Final Answer:

   spark.sql.adaptive.enabled = true -> Option D

4. Quick Check:

   Enable adaptive query with spark.sql.adaptive.enabled [OK]

Hint: Remember full key: spark.sql.adaptive.enabled [OK]

Common Mistakes:

• Using incorrect config keys like spark.execution.adaptive
• Setting the value to false instead of true
• Mixing up config names with similar words

3.

Given the following Spark SQL code with adaptive query execution enabled, what will be the effect on the join?

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

val df1 = spark.range(1000)
val df2 = spark.range(10)

val joined = df1.join(df2, df1("id") === df2("id"))
joined.explain(true)

What does adaptive query execution do here?

medium

A. It converts the join to a broadcast join because df2 is small

B. It disables the join optimization and runs a shuffle join

C. It throws an error because adaptive execution cannot handle joins

D. It caches both dataframes before joining

Solution

1. Step 1: Identify adaptive execution behavior on join

   Adaptive query execution detects small tables and converts joins to broadcast joins to reduce shuffle.

2. Step 2: Analyze data sizes in example

   df2 has only 10 rows, which is small enough to broadcast.

3. Final Answer:

   It converts the join to a broadcast join because df2 is small -> Option A

4. Quick Check:

   Small table join = broadcast join [OK]

Hint: Small table in join? Adaptive broadcasts it [OK]

Common Mistakes:

• Assuming adaptive disables optimization
• Thinking adaptive causes errors on joins
• Believing it caches dataframes automatically

4.

Consider this Spark code snippet with adaptive query execution enabled:

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

val df1 = spark.range(1000)
val df2 = spark.range(1000000)

val joined = df1.join(df2, df1("id") === df2("id"))
joined.show()

The job runs very slowly. What is the likely cause?

medium

A. Adaptive query execution is disabled by default

B. Adaptive query execution tries to broadcast df2, which is too large

C. The join condition is invalid causing a runtime error

D. Spark automatically caches df2 causing memory overflow

Solution

1. Step 1: Understand adaptive join behavior with large tables

   Adaptive query execution may try to broadcast a table if it thinks it's small enough.

2. Step 2: Analyze table sizes

   df2 has 1,000,000 rows, which is too large to broadcast efficiently, causing slow execution.

3. Final Answer:

   Adaptive query execution tries to broadcast df2, which is too large -> Option B

4. Quick Check:

   Broadcasting large tables causes slow jobs [OK]

Hint: Large table broadcast causes slow join [OK]

Common Mistakes:

• Assuming adaptive is disabled by default
• Thinking join condition is invalid
• Believing Spark caches large tables automatically

5.

You have a Spark job joining three datasets of unknown sizes. How can adaptive query execution help optimize this multi-join query?

hard

A. It forces all joins to be shuffle joins regardless of size

B. It disables all joins and runs map-only jobs

C. It dynamically chooses broadcast or shuffle joins for each join based on runtime data size

D. It requires manual hints to optimize each join

Solution

1. Step 1: Understand adaptive query execution with multiple joins

   Adaptive query execution evaluates data sizes during runtime for each join.

2. Step 2: Explain optimization choice

   It chooses broadcast joins for small tables and shuffle joins for large tables automatically.

3. Step 3: Apply to multi-join scenario

   For three datasets, it optimizes each join individually based on actual data size.

4. Final Answer:

   It dynamically chooses broadcast or shuffle joins for each join based on runtime data size -> Option C

5. Quick Check:

   Adaptive = dynamic join strategy per join [OK]

Hint: Adaptive picks join type per join at runtime [OK]

Common Mistakes:

• Thinking adaptive disables joins
• Assuming all joins become shuffle joins
• Believing manual hints are always required