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.
Contents
- 1 Key Features of AQE
- 2 1. Dynamic Partition Pruning
- 3 2. Reoptimization Based on Runtime Statistics
- 4 3. Dynamic Coalescing of Shuffle Partitions
- 5 4. Dynamic Join Reordering and Selection
- 6 Enabling AQE
- 7 Configuration Properties
- 8 Example
- 9 Benefits of AQE
- 10 Example of How Adaptive Query Execution (AQE) Optimizes Queries in Apache Spark
- 11 Scenario Setup
- 12 Without AQE
- 13 With AQE
- 14 AQE Optimizations in Action
- 15 Visualization of Optimizations
- 16 Benefits Observed
- 17 Enabling AQE
- 18 Additional AQE Configuration Options
- 19 Complete Example
- 20 Share this:
Key Features of AQE
- Dynamic Partition Pruning
- Reoptimization Based on Runtime Statistics
- Dynamic Coalescing of Shuffle Partitions
- 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
- Improved Performance: By dynamically adjusting the execution plan based on actual data characteristics, AQE can significantly improve query performance.
- Better Resource Utilization: AQE helps in avoiding inefficient execution plans, leading to better resource utilization.
- 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:
- Fixed Number of Shuffle Partitions: Spark may use a fixed number of shuffle partitions, leading to inefficient processing if data size varies significantly.
- Suboptimal Join Strategy: Spark may not choose the most efficient join strategy due to lack of runtime statistics.
- 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
- Dynamic Partition Pruning:AQE can prune partitions dynamically based on the filter condition applied to the
sales
table. Ifsales.amount > 1000
eliminates some partitions, those partitions won’t be read from thecustomers
table during the join. - 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. - 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.
- 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 thesales
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
- Reduced Data Read: With dynamic partition pruning, only the necessary partitions are read, reducing I/O operations.
- Efficient Task Execution: Coalescing shuffle partitions results in fewer, more efficient tasks.
- Balanced Execution: Handling skew joins results in balanced task distribution, avoiding bottlenecks.
- 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 istrue
.
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 istrue
.
spark = SparkSession.builder \ .appName("Spark AQE Example") \ .config("spark.sql.adaptive.enabled", "true") \ .config("spark.sql.adaptive.join.enabled", "true") \ .getOrCreate()
Handling Skew Joinsspark.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 is1
.spark.sql.adaptive.coalescePartitions.maxPartitionNum
: Maximum number of coalesced partitions. Default isInteger.MAX_VALUE
.
spark = 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 isINFO
.
spark = 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.
Leave a Reply