Adaptive Query Execution (AQE) in Apache Spark 3.0 is a powerful feature that brings more intelligent and dynamic optimizations to Spark SQL on runtime statistics. By adapting the execution plan at runtime based on actual data statistics, AQE can provide significant performance improvements and more efficient resource utilization. Enabling AQE is straightforward, and it can bring substantial benefits to big data processing workloads.

Key Features of AQE

1. Dynamic Partition Pruning
2. Reoptimization Based on Runtime Statistics
3. Dynamic Coalescing of Shuffle Partitions
4. Dynamic Join Reordering and Selection

1. Dynamic Partition Pruning

Dynamic partition pruning helps in reducing the amount of data read during a join operation by pruning unnecessary partitions. This optimization occurs at runtime when the actual partition data is known.

How it Works

• When joining two tables, Spark can dynamically prune partitions of one table based on the filter conditions applied to the other table.
• This is especially beneficial for star-schema queries common in data warehousing.

2. Reoptimization Based on Runtime Statistics

Reoptimization allows Spark to adjust the execution plan based on runtime statistics, such as the size of intermediate data. This can help in choosing better join strategies or avoiding skewed data distribution.

How it Works

• Spark collects statistics during the execution of stages.
• Based on these statistics, Spark can modify the execution plan, such as changing the join type or repartitioning data.

3. Dynamic Coalescing of Shuffle Partitions

Dynamic coalescing reduces the number of shuffle partitions based on the actual data size, which helps in avoiding small, inefficient tasks.

How it Works

• Spark initially uses a large number of shuffle partitions.
• During the shuffle, it measures the size of the data and dynamically coalesces (combines) small partitions into larger ones.

4. Dynamic Join Reordering and Selection

Dynamic join reordering and selection optimize the order of join operations and the join strategies based on the size of the data at runtime.

How it Works

• Spark collects size estimates of the intermediate data.
• Based on these estimates, Spark can reorder the joins to process smaller tables first or switch to a more efficient join strategy.

Enabling AQE

To enable AQE in Spark, you need to set the configuration properties in your Spark session:

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("Spark AQE Example") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .config("spark.sql.adaptive.join.enabled", "true") \
    .getOrCreate()

Configuration Properties

• spark.sql.adaptive.enabled: Enables AQE (default: false).
• spark.sql.adaptive.coalescePartitions.enabled: Enables coalescing shuffle partitions (default: true).
• spark.sql.adaptive.skewJoin.enabled: Enables skew join optimization (default: true).
• spark.sql.adaptive.join.enabled: Enables join reordering (default: true).

Example

Let’s consider a scenario where we have two tables, large_table and small_table, and we want to perform a join operation.

pythonCopy codefrom pyspark.sql import SparkSession

# Initialize Spark session with AQE enabled
spark = SparkSession.builder \
    .appName("Spark AQE Example") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

# Load the tables
large_table = spark.read.parquet("hdfs:///path/to/large_table")
small_table = spark.read.parquet("hdfs:///path/to/small_table")

# Perform join operation
result = large_table.join(small_table, "key")

# Show result
result.show()

Benefits of AQE

1. Improved Performance: By dynamically adjusting the execution plan based on actual data characteristics, AQE can significantly improve query performance.
2. Better Resource Utilization: AQE helps in avoiding inefficient execution plans, leading to better resource utilization.
3. Reduced Data Shuffling: Dynamic partition pruning and coalescing of shuffle partitions help in reducing the amount of data shuffling, which is often a bottleneck in big data processing.

Example of How Adaptive Query Execution (AQE) Optimizes Queries in Apache Spark

To illustrate how AQE optimizes queries, let’s consider an example where we have two tables: sales and customers. We’ll perform a join operation between these tables and see how AQE improves the execution.

Scenario Setup

Let’s assume:

• sales table has millions of rows representing sales transactions.
• customers table has thousands of rows representing customer details.
• We want to join these tables on the customer_id column and filter the results based on a condition.

Without AQE

Without AQE, Spark may not have the most optimal execution plan. Here’s how a typical query execution might look:

from pyspark.sql import SparkSession

# Initialize Spark session without AQE
spark = SparkSession.builder \
    .appName("Spark AQE Example Without AQE") \
    .config("spark.sql.adaptive.enabled", "false") \
    .getOrCreate()

# Load the tables
sales = spark.read.parquet("hdfs:///path/to/sales")
customers = spark.read.parquet("hdfs:///path/to/customers")

# Perform join operation
result = sales.join(customers, "customer_id") \
              .filter(sales.amount > 1000)

# Show result
result.show()

Issues without AQE:

1. Fixed Number of Shuffle Partitions: Spark may use a fixed number of shuffle partitions, leading to inefficient processing if data size varies significantly.
2. Suboptimal Join Strategy: Spark may not choose the most efficient join strategy due to lack of runtime statistics.
3. No Dynamic Partition Pruning: All partitions are read even if some are not needed.

With AQE

Now, let’s enable AQE and see how it optimizes the query execution.

from pyspark.sql import SparkSession

# Initialize Spark session with AQE
spark = SparkSession.builder \
    .appName("Spark AQE Example With AQE") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .config("spark.sql.adaptive.join.enabled", "true") \
    .getOrCreate()

# Load the tables
sales = spark.read.parquet("hdfs:///path/to/sales")
customers = spark.read.parquet("hdfs:///path/to/customers")

# Perform join operation
result = sales.join(customers, "customer_id") \
              .filter(sales.amount > 1000)

# Show result
result.show()

AQE Optimizations in Action

1. Dynamic Partition Pruning:AQE can prune partitions dynamically based on the filter condition applied to the sales table. If sales.amount > 1000 eliminates some partitions, those partitions won’t be read from the customers table during the join.
2. Dynamic Coalescing of Shuffle Partitions:AQE dynamically adjusts the number of shuffle partitions based on the actual size of the data. If the sales table has uneven distribution of data, AQE can coalesce small partitions into larger ones to avoid creating many small, inefficient tasks.
3. Skew Join Handling:AQE detects skewed data during the join operation and can adjust the join strategy to handle the skew, leading to more balanced and efficient execution.
4. Reoptimization Based on Runtime Statistics:AQE collects runtime statistics and can reoptimize the join strategy. For example, if the customers table is much smaller than the sales table, AQE might choose a broadcast join to speed up the operation.

Visualization of Optimizations

Let’s compare the execution plans with and without AQE.

Without AQE

result.explain(True)

Output (simplified):

== Physical Plan ==
*(2) BroadcastHashJoin [customer_id#5L], [customer_id#10L], Inner, BuildRight
:- *(2) Project [customer_id#5L, amount#7]
:  +- *(2) Filter (amount#7 > 1000)
:     +- *(2) FileScan parquet [customer_id#5L, amount#7] Batched: true, Format: Parquet
+- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, true]))
   +- *(1) FileScan parquet [customer_id#10L] Batched: true, Format: Parquet

With AQE

result.explain(True)

Output (simplified):

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- *(2) BroadcastHashJoin [customer_id#5L], [customer_id#10L], Inner, BuildRight
   :- *(2) Project [customer_id#5L, amount#7]
   :  +- *(2) Filter (amount#7 > 1000)
   :     +- *(2) FileScan parquet [customer_id#5L, amount#7] Batched: true, Format: Parquet
   +- BroadcastExchange HashedRelationBroadcastMode(List(input[0, bigint, true]))
      +- *(1) FileScan parquet [customer_id#10L] Batched: true, Format: Parquet

Benefits Observed

1. Reduced Data Read: With dynamic partition pruning, only the necessary partitions are read, reducing I/O operations.
2. Efficient Task Execution: Coalescing shuffle partitions results in fewer, more efficient tasks.
3. Balanced Execution: Handling skew joins results in balanced task distribution, avoiding bottlenecks.
4. Optimal Join Strategy: Runtime statistics help in choosing the best join strategy, improving performance.

Adaptive Query Execution (AQE) significantly enhances the efficiency and performance of Spark SQL jobs by dynamically adjusting the execution plan based on runtime data statistics. By enabling AQE, users can leverage optimizations like dynamic partition pruning, coalescing shuffle partitions, handling skew joins, and reoptimization based on runtime statistics to achieve better resource utilization and faster query execution.

Enabling Adaptive Query Execution (AQE) in Spark is straightforward and involves setting several configuration properties in your Spark session. Here are the key configurations you need to enable AQE and some additional options to fine-tune its behavior:

Enabling AQE

To enable AQE, you need to set the spark.sql.adaptive.enabled property to true. This enables the core functionality of AQE. Here’s how you can do it in your Spark session:

pythonCopy codefrom pyspark.sql import SparkSession

# Initialize Spark session with AQE enabled
spark = SparkSession.builder \
    .appName("Spark AQE Example") \
    .config("spark.sql.adaptive.enabled", "true") \
    .getOrCreate()

Additional AQE Configuration Options

Dynamic Coalescing of Shuffle Partitions

• spark.sql.adaptive.coalescePartitions.enabled: Enables or disables dynamic coalescing of shuffle partitions. Default is true.

spark = SparkSession.builder \ .appName("Spark AQE Example") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \ .getOrCreate()

Dynamic Join Reordering and Selection

• spark.sql.adaptive.join.enabled: Enables or disables dynamic join reordering and selection. Default is true.

pythonCopy codespark = SparkSession.builder \ .appName("Spark AQE Example") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.join.enabled", "true") \ .getOrCreate()

Handling Skew Joins
spark.sql.adaptive.skewJoin.enabled: Enables or disables skew join handling. Default is true
spark = SparkSession.builder \ .appName("Spark AQE Example") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.skewJoin.enabled", "true") \ .getOrCreate()

Minimum and Maximum Number of Coalesced Partitions

• spark.sql.adaptive.coalescePartitions.minPartitionNum: Minimum number of coalesced partitions. Default is 1.
• spark.sql.adaptive.coalescePartitions.maxPartitionNum: Maximum number of coalesced partitions. Default is Integer.MAX_VALUE.

pythonCopy codespark = SparkSession.builder \ .appName("Spark AQE Example") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.coalescePartitions.minPartitionNum", "2") \ .config("spark.sql.adaptive.coalescePartitions.maxPartitionNum", "1000") \ .getOrCreate()

Adaptive Plan Changes Logging

• spark.sql.adaptive.logLevel: Log level for adaptive plan changes. Default is INFO.

pythonCopy codespark = SparkSession.builder \ .appName("Spark AQE Example") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.logLevel", "DEBUG") \ .getOrCreate()

Other Useful AQE Properties

spark.sql.adaptive.advisoryPartitionSizeInBytes: The target size for coalesced shuffle partitions. Default is 64MB.

spark.sql.adaptive.skewJoin.skewedPartitionFactor: The skew factor to determine whether a partition is skewed. Default is 1.5.

spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes: The threshold in bytes to consider a partition as skewed. Default is 256MB.

Complete Example

Here is a complete example showing how to enable AQE with various configuration options:

from pyspark.sql import SparkSession

# Initialize Spark session with AQE enabled and other configurations
spark = SparkSession.builder \
    .appName("Spark AQE Example") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .config("spark.sql.adaptive.join.enabled", "true") \
    .config("spark.sql.adaptive.skewJoin.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.minPartitionNum", "2") \
    .config("spark.sql.adaptive.coalescePartitions.maxPartitionNum", "1000") \
    .config("spark.sql.adaptive.logLevel", "DEBUG") \
    .config("spark.sql.adaptive.advisoryPartitionSizeInBytes", "64MB") \
    .config("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "1.5") \
    .config("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB") \
    .getOrCreate()

# Load the tables
sales = spark.read.parquet("hdfs:///path/to/sales")
customers = spark.read.parquet("hdfs:///path/to/customers")

# Perform join operation
result = sales.join(customers, "customer_id") \
              .filter(sales.amount > 1000)

# Show result
result.show()

Enabling Adaptive Query Execution (AQE) in Spark involves setting the spark.sql.adaptive.enabled property to true along with other optional configurations to fine-tune its behavior. AQE dynamically adjusts query plans based on runtime statistics, which can significantly improve performance and resource utilization for Spark SQL queries.