Adaptive query execution in Apache Spark - Practice Problems & Coding Challenges

Challenge - 5 Problems

🧠 Conceptual

intermediate

What is the main benefit of Adaptive Query Execution (AQE) in Apache Spark?

Adaptive Query Execution (AQE) dynamically adjusts query plans based on runtime statistics. What is the primary benefit of using AQE?

A It disables all shuffle operations to speed up queries.

B It guarantees that queries always run faster than non-AQE queries.

C It reduces the need for manual query tuning by optimizing plans during execution.

D It converts all joins into broadcast joins regardless of data size.

💡 Hint

Think about how AQE helps with query optimization without manual intervention.

AQE improves query performance by dynamically optimizing the query plan during execution using runtime statistics, reducing manual tuning effort.

❓ Predict Output

intermediate

Output of AQE enabled join size adjustment

Given the following Spark code with AQE enabled, what will be the number of partitions after the join?

Apache Spark

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

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

# Simulate two DataFrames
left = spark.range(0, 1000)
right = spark.range(0, 10)

joined = left.join(right, left.id == right.id)

partitions = joined.rdd.getNumPartitions()
print(partitions)

A 10

B 200

C 1

D 1000

💡 Hint

Consider how AQE optimizes join partitioning when one side is small.

With AQE enabled, Spark detects the small size of the right DataFrame and reduces the number of partitions to match the small side, here 10.

🔧 Debug

advanced

Identify the error caused by disabling AQE incorrectly

What error will occur if you disable AQE but still use AQE-specific configuration options in Spark?

Apache Spark

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

# Run a query that triggers coalescePartitions
spark.range(1000).repartition(10).count()

A Spark silently enables AQE despite the setting.

B RuntimeException: AQE is disabled but coalescePartitions is enabled.

C AnalysisException: Invalid configuration combination.

D No error, query runs normally ignoring coalescePartitions.

💡 Hint

Think about how Spark handles AQE-related configs when AQE is off.

When AQE is disabled, AQE-specific configs like coalescePartitions are ignored without error, so the query runs normally.

❓ data_output

advanced

Resulting shuffle partitions after AQE coalesce

Given a Spark job with 100 shuffle partitions and AQE coalesce enabled, if the actual shuffle data size is small, what will be the number of shuffle partitions after AQE coalescing?

Apache Spark

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.shuffle.partitions", "100")

# Simulate a small shuffle by filtering
df = spark.range(1000).filter("id < 10")
df2 = df.repartition(100)

partitions_before = df2.rdd.getNumPartitions()

# Trigger an action to run AQE
df2.count()

partitions_after = df2.rdd.getNumPartitions()
print(partitions_after)

A 100

B 10

C 50

D 1

💡 Hint

Consider when AQE coalesce partitions takes effect.

AQE coalesce partitions happens during shuffle stage optimization, but repartition creates a new RDD with fixed partitions. The number of partitions remains 100 here.

🚀 Application

expert

Choosing the best AQE setting for skewed data joins

You have two large DataFrames with skewed keys joining on a column. Which AQE setting helps reduce the impact of skew by splitting large partitions dynamically?

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

B spark.sql.adaptive.skewJoin.enabled = true

C spark.sql.adaptive.enabled = false

D spark.sql.shuffle.partitions = 1

💡 Hint

Think about AQE features that handle skewed joins.

Enabling spark.sql.adaptive.skewJoin.enabled allows AQE to detect skewed partitions and split them to balance workload.

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