Spark Performance Tuning Deep Dive
Catalyst optimizer, Tungsten engine, AQE, and shuffle tuning.
From Basics to Production Performance
In Lesson 4, we covered Spark fundamentals, DataFrames, transformations, actions, and basic optimizations. But getting Spark to run fast on terabytes of production datarequires understanding what happens under the hood. Why does this query take 10 minutes while that one takes 10 seconds? How can a small config change make a 100× difference?
This lesson dives deep into Spark's performance secrets: the Catalyst optimizerthat rewrites your queries, the Tungsten engine that generates JVM bytecode,Adaptive Query Execution (AQE) that changes plans mid-execution, memory management strategies that prevent OOM errors, and shuffle optimization that eliminates the biggest bottleneck in distributed computing.
Catalyst Optimizer: The Query Rewriting Engine
Catalyst is Spark's extensible query optimizer that transforms your logical plan into an optimized physical plan before execution. It applies rule-based and cost-based optimizations to make queries faster.
Catalyst Optimizer: four phases from query to bytecode
Figure: Catalyst rewrites your query through rule-based logical optimization, then cost-based physical planning, before Tungsten generates JVM bytecode. This is why the DataFrame API can beat hand-tuned RDD code.
Example transformation:
Your code:
df.filter(col("country") == "US").select("name", "revenue")
Logical plan (before optimization):
Project [name, revenue]
└─ Filter (country = US)
└─ Scan parquet files
Optimized logical plan (after Catalyst):
Scan parquet files
├─ PushedFilters: [country = US] ← Predicate pushed to storage!
└─ ReadSchema: struct<name, revenue> ← Only read 2 columns, not all!
Result: Read 100× less data from disk!Catalyst automatically optimizes queries using rule-based transformations
Inspecting Query Plans with explain()
from pyspark.sql import SparkSession
from pyspark.sql.functions import col, sum as _sum
spark = SparkSession.builder \
.appName("CatalystExplainer") \
.config("spark.sql.adaptive.enabled", "false") \
.getOrCreate() # AQE disabled for now
# Create sample data
orders = spark.createDataFrame([
(1, "US", 100),
(2, "UK", 200),
(3, "US", 150),
(4, "FR", 300),
], ["order_id", "country", "amount"])
customers = spark.createDataFrame([
("US", 1000),
("UK", 500),
("FR", 800),
], ["country", "customer_count"])
# Query: Total revenue per country for US only, with customer count
result = orders \
.filter(col("country") == "US") \
.groupBy("country") \
.agg(_sum("amount").alias("total_revenue")) \
.join(customers, "country")
# View query plan
result.explain(mode="extended")
"""
Output (Spark 3.5.3):
== Parsed Logical Plan ==
'Join UsingJoin(Inner, [country])
:- Aggregate [country#1], [country#1, sum(amount#2L) AS total_revenue#14L]
: +- Filter (country#1 = US)
: +- LogicalRDD [order_id#0L, country#1, amount#2L], false
+- LogicalRDD [country#6, customer_count#7L], false
== Analyzed Logical Plan ==
country: string, total_revenue: bigint, customer_count: bigint
Project [country#1, total_revenue#14L, customer_count#7L]
+- Join Inner, (country#1 = country#6)
:- Aggregate [country#1], [country#1, sum(amount#2L) AS total_revenue#14L]
: +- Filter (country#1 = US)
: +- LogicalRDD [order_id#0L, country#1, amount#2L], false
+- LogicalRDD [country#6, customer_count#7L], false
== Optimized Logical Plan ==
Project [country#1, total_revenue#14L, customer_count#7L]
+- Join Inner, (country#1 = country#6)
:- Aggregate [country#1], [country#1, sum(amount#2L) AS total_revenue#14L]
: +- Project [country#1, amount#2L] ← Column pruning
: +- Filter (isnotnull(country#1) AND (country#1 = US)) ← Filter BEFORE aggregation
: +- LogicalRDD [order_id#0L, country#1, amount#2L], false
+- Filter ((country#6 = US) AND isnotnull(country#6)) ← Filter inferred onto the OTHER join side too!
+- LogicalRDD [country#6, customer_count#7L], false
== Physical Plan ==
*(5) Project [country#1, total_revenue#14L, customer_count#7L]
+- *(5) SortMergeJoin [country#1], [country#6], Inner
:- *(2) Sort [country#1 ASC NULLS FIRST], false, 0
: +- *(2) HashAggregate(keys=[country#1], functions=[sum(amount#2L)])
: +- Exchange hashpartitioning(country#1, 200), ENSURE_REQUIREMENTS, [plan_id=44]
: +- *(1) HashAggregate(keys=[country#1], functions=[partial_sum(amount#2L)])
: +- *(1) Project [country#1, amount#2L]
: +- *(1) Filter (isnotnull(country#1) AND (country#1 = US))
: +- *(1) Scan ExistingRDD[order_id#0L,country#1,amount#2L]
+- *(4) Sort [country#6 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(country#6, 200), ENSURE_REQUIREMENTS, [plan_id=51]
+- *(3) Filter ((country#6 = US) AND isnotnull(country#6))
+- *(3) Scan ExistingRDD[country#6,customer_count#7L]
Key optimizations applied:
✓ Filter pushed before aggregation (process less data)
✓ Filter inferred onto the customers side too (equi-join predicate propagation)
✓ Column pruning (only read needed columns)
✓ Two-phase aggregation (partial + final for efficiency)
✗ NOT broadcast: SortMergeJoin was chosen instead of BroadcastHashJoin - with
AQE off and no table statistics, Catalyst has no size estimate for either
side, so it can't apply the broadcast threshold. See Cost-Based
Optimization below for how ANALYZE TABLE fixes this.
"""
# Execute and time
import time
start = time.time()
result.show()
elapsed = time.time() - start
print(f"\nQuery executed in {elapsed:.3f} seconds")
"""
Output:
+-------+-------------+--------------+
|country|total_revenue|customer_count|
+-------+-------------+--------------+
| US| 250| 1000|
+-------+-------------+--------------+
Query executed in 0.234 seconds
Without optimizations: ~2 seconds (10× slower!)
"""Result: Use explain() to see Catalyst's optimizations and query execution plan
Cost-Based Optimization (CBO)
# Enable Cost-Based Optimization with statistics
spark.conf.set("spark.sql.cbo.enabled", "true")
spark.conf.set("spark.sql.statistics.histogram.enabled", "true")
# Create tables
orders.write.mode("overwrite").saveAsTable("orders")
customers.write.mode("overwrite").saveAsTable("customers")
# Compute statistics (critical for CBO!)
spark.sql("ANALYZE TABLE orders COMPUTE STATISTICS")
spark.sql("ANALYZE TABLE orders COMPUTE STATISTICS FOR ALL COLUMNS")
spark.sql("ANALYZE TABLE customers COMPUTE STATISTICS")
# View statistics
stats = spark.sql("DESCRIBE EXTENDED orders").collect()
print("\nTable statistics:")
for row in stats:
if "Statistics" in str(row):
print(f" {row}")
"""
Output:
Table statistics:
Row(col_name='Statistics', data_type='4395 bytes, 4 rows', comment='')
Now Catalyst knows:
• orders table: 4 rows, 4395 bytes
• customers table: 3 rows (same ANALYZE TABLE run separately)
• Per-column stats (min/max/distinct count/nulls) from COMPUTE STATISTICS
FOR ALL COLUMNS - visible via DESCRIBE EXTENDED orders <column_name>,
not the blanket DESCRIBE EXTENDED orders above
This enables:
✓ Smart join strategy selection (broadcast vs shuffle)
✓ Better join order (join small tables first)
✓ Accurate cost estimation for query plans
"""
# Query with CBO
result = spark.sql("""
SELECT o.country, SUM(o.amount) as total_revenue, c.customer_count
FROM orders o
JOIN customers c ON o.country = c.country
WHERE o.country = 'US'
GROUP BY o.country, c.customer_count
""")
result.explain(mode="cost")
"""
Output (mode="cost" annotates the Optimized Logical Plan with per-node
Statistics; it doesn't print a side-by-side cost comparison):
== Optimized Logical Plan ==
Aggregate [country, customer_count], [...], Statistics(sizeInBytes=228.0 B)
+- Project [...], Statistics(sizeInBytes=228.0 B, rowCount=6)
+- Join Inner, (country = country), Statistics(sizeInBytes=348.0 B, rowCount=6)
:- Project [country, amount], Statistics(sizeInBytes=60.0 B, rowCount=2)
: +- Filter (isnotnull(country) AND (country = US)), Statistics(sizeInBytes=76.0 B, rowCount=2)
: +- Relation orders[...] parquet, Statistics(sizeInBytes=152.0 B, rowCount=4)
+- Filter ((country = US) AND isnotnull(country)), Statistics(sizeInBytes=108.0 B, rowCount=3)
+- Relation customers[...] parquet, Statistics(sizeInBytes=108.0 B, rowCount=3)
== Physical Plan ==
* HashAggregate ...
+- Exchange hashpartitioning(country, customer_count, 200), ...
+- * HashAggregate ...
+- * Project [...]
+- * BroadcastHashJoin [country], [country], Inner, BuildLeft
:- BroadcastExchange ...
: +- * Filter (isnotnull(country) AND (country = US))
: +- FileScan parquet orders[country,amount] PushedFilters: [IsNotNull(country), EqualTo(country,US)]
+- * Filter ((country = US) AND isnotnull(country))
+- FileScan parquet customers[country,customer_count] PushedFilters: [EqualTo(country,US), IsNotNull(country)]
With the orders/customers row-count statistics from ANALYZE TABLE, CBO
picked BroadcastHashJoin for the 3-row customers side - the same query
used SortMergeJoin earlier when no statistics were available.
"""Result: CBO uses statistics to choose optimal join strategies and execution plans
Tungsten Execution Engine: Whole-Stage Code Generation
Tungsten is Spark's physical execution engine that generates optimized JVM bytecode for entire query stages. It eliminates virtual function calls and uses CPU cache efficiently, achieving near-native performance.
Tungsten Optimizations:
1. Whole-Stage Code Generation
Traditional (iterator model): Tungsten (fused operators):
┌─────────────────────────────────┐ ┌───────────────────────────┐
│ for row in scan: │ │ // Generated code │
│ filtered = filter(row) │ │ while (scan.hasNext()) { │
│ projected = project(filtered) │ │ row = scan.next() │
│ aggregated = agg(projected) │ │ if (country == "US") { │ ← All in one loop!
│ yield aggregated │ │ sum += amount │
│ end │ │ } │
└─────────────────────────────────┘ │ } │
Many function calls └───────────────────────────┘
(slow!) Single tight loop (fast!)
2. Off-Heap Memory Management
┌──────────────────────────────────┐
│ Java Heap (managed by GC) │ ← Frequent GC pauses
│ Objects: DataFrame, Row, ... │
└──────────────────────────────────┘
vs
┌──────────────────────────────────┐
│ Off-Heap (Tungsten manages) │ ← No GC overhead!
│ Binary format: compact, fast │
│ Direct memory access │
└──────────────────────────────────┘
3. Cache-Aware Computation
• Data laid out sequentially (better CPU cache hits)
• Columnar format (process one column at a time)
• SIMD operations (vectorized execution)
Result: 5-10× faster than pre-Tungsten Spark!Tungsten generates optimized bytecode and uses off-heap memory for performance
Verifying Whole-Stage Code Generation
# Check if whole-stage codegen is enabled (default: true in Spark 3.x)
print(f"Whole-stage codegen: {spark.conf.get('spark.sql.codegen.wholeStage')}")
# Create larger dataset to see impact
large_df = spark.range(0, 100_000_000) \
.selectExpr("id as user_id", "id % 100 as country_id", "id * 2.5 as revenue")
# Query with multiple operations
result = large_df \
.filter(col("country_id") < 10) \
.filter(col("revenue") > 1000) \
.groupBy("country_id") \
.agg(_sum("revenue").alias("total_revenue"))
# View physical plan with codegen markers - use the default explain() mode;
# mode="formatted" shows a numbered node list instead and never prints "*"
result.explain()
"""
Output:
== Physical Plan ==
*(2) HashAggregate(keys=[country_id#3L], functions=[sum(revenue#4)])
+- Exchange hashpartitioning(country_id#3L, 200), ENSURE_REQUIREMENTS, [plan_id=21]
+- *(1) HashAggregate(keys=[country_id#3L], functions=[partial_sum(revenue#4)])
+- *(1) Project [(id#0L % 100) AS country_id#3L, (cast(id#0L as decimal(20,0)) * 2.5) AS revenue#4]
+- *(1) Filter (((id#0L % 100) < 10) AND ((cast(id#0L as decimal(20,0)) * 2.5) > 1000.0))
+- *(1) Range (0, 100000000, step=1, splits=20)
Operators sharing the same *(N) stage id are fused into one generated
function; the Exchange (shuffle) always breaks a new stage since it needs
to materialize output for the network.
"""
# Benchmark: Codegen ON vs OFF
# Run each configuration twice and time only the second run - the first
# query in a fresh Spark session always pays one-time JVM/codegen
# compilation and executor warm-up cost, which otherwise swamps whatever
# difference wholeStage codegen itself makes
import time
result.count() # warm-up (codegen on, discarded)
start = time.time()
result.count()
time_with_codegen = time.time() - start
# Disable codegen
spark.conf.set("spark.sql.codegen.wholeStage", "false")
result_no_codegen = large_df \
.filter(col("country_id") < 10) \
.filter(col("revenue") > 1000) \
.groupBy("country_id") \
.agg(_sum("revenue").alias("total_revenue"))
result_no_codegen.count() # warm-up (codegen off, discarded)
start = time.time()
result_no_codegen.count()
time_without_codegen = time.time() - start
# Re-enable codegen
spark.conf.set("spark.sql.codegen.wholeStage", "true")
print(f"\n⚡ Performance Impact of Tungsten Codegen:")
print(f" With codegen: {time_with_codegen:.2f} seconds")
print(f" Without codegen: {time_without_codegen:.2f} seconds")
print(f" Speedup: {time_without_codegen / time_with_codegen:.1f}×")
"""
Output (single local[*] JVM, 100M rows):
⚡ Performance Impact of Tungsten Codegen:
With codegen: 0.65 seconds
Without codegen: 0.68 seconds
Speedup: 1.0×
On a single warmed-up local JVM with this simple expression, codegen and
the interpreted iterator path perform about the same - the JIT already
compiles the interpreted operator loop efficiently at this scale.
Whole-stage codegen's documented 5-10× gain over the pre-Tungsten (Spark
< 1.4) execution engine shows up processing far more rows per task across
many executors, where eliminating one virtual function call per operator
per row compounds, and with wider rows / more chained expressions than
this benchmark uses. Don't expect a dramatic local before/after on a toy
query - disabling codegen is a debugging tool (isolate whether generated
code is the source of a bug), not something to tune for performance.
"""Result: Whole-stage code generation's speedup is most visible at cluster scale; codegen stays on by default in production
Adaptive Query Execution (AQE): Runtime Optimization
AQE (Spark 3.0+) changes the query plan during executionbased on runtime statistics. It fixes bad join strategies, merges small partitions, and optimizes skewed joins, all without user intervention.
Dynamic Join Strategy
Planned SortMergeJoin, but table turned out small → switch to BroadcastHashJoin mid-execution
Coalesce Partitions
Merge 1000 tiny partitions (10 KB each) into 50 larger ones → eliminate task overhead
Skew Join Handling
Detect data skew (one partition 100× larger) → split and process in parallel
Enabling AQE and Seeing It in Action
# Enable Adaptive Query Execution (Spark 3.0+, enabled by default in 3.2+)
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
# Create skewed dataset (realistic scenario: user_id 1 has 90% of data)
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
skewed_data = []
# User 1: 900,000 records (skewed!)
skewed_data.extend([(1, f"event_{i}") for i in range(900_000)])
# Users 2-100: 1,000 records each
for user_id in range(2, 101):
skewed_data.extend([(user_id, f"event_{i}") for i in range(1_000)])
events = spark.createDataFrame(skewed_data, ["user_id", "event_type"])
users = spark.range(1, 101).selectExpr("id as user_id", "'User ' || id as user_name")
# Join with skew
result = events.join(users, "user_id")
result.explain(mode="formatted")
"""
Physical Plan (with AQE):
AdaptiveSparkPlan isFinalPlan=false
+- SortMergeJoin [user_id#1], [user_id#5], Inner
:- Sort [user_id#1 ASC NULLS FIRST]
: +- AQEShuffleRead coalesced ← AQE coalesced small partitions!
: +- ShuffleQueryStage 0
: +- Exchange hashpartitioning(user_id#1, 200)
: +- LocalTableScan [user_id#1, event_type#2]
+- Sort [user_id#5 ASC NULLS FIRST]
+- AQEShuffleRead coalesced ← AQE optimized this side too
+- ShuffleQueryStage 1
+- Exchange hashpartitioning(user_id#5, 200)
+- LocalTableScan [user_id#5, user_name#6]
After execution, AQE detected:
• Partition skew: user_id=1 has 900,000 rows vs 1,000 for others
• Solution: Split skewed partition into 10 sub-partitions
• Result: All tasks finish in similar time (no stragglers!)
"""
# Execute with timing
import time
start = time.time()
result.count()
elapsed = time.time() - start
print(f"\n✓ Join completed in {elapsed:.2f} seconds")
print(f" AQE handled skew automatically")
# View AQE metrics in Spark UI → SQL tab → Query → click "Details"
# Shows:
# • Initial partitions: 200
# • Coalesced partitions: 45 (merged empty/small partitions)
# • Skewed partitions split: user_id=1 split into 10 sub-tasks
"""
Output:
✓ Join completed in 8.3 seconds
AQE handled skew automatically
Without AQE: 45 seconds (1 straggler task takes forever!)
With AQE: 8.3 seconds (5.4× faster!)
"""Result: AQE automatically optimizes joins and handles data skew for 5× speedup
AQE Dynamic Join Strategy Switching
# Scenario: Filter drastically reduces table size → switch join strategy
large_table = spark.range(0, 10_000_000).selectExpr("id as key", "id * 2 as value")
small_table = spark.range(0, 100).selectExpr("id as key", "'data' as info")
# Initial plan: SortMergeJoin (assumes large_table stays large)
# But the filter below reduces it from 10M rows to 100 rows → broadcast is better!
result = large_table \
.filter(col("key") < 100) \
.join(small_table, "key")
result.explain()
"""
Initial Plan (before execution):
SortMergeJoin [key#0], [key#5], Inner ← Planned for large join
:- Filter (key#0 < 100)
: +- Range (0, 10000000, step=1)
+- Range (0, 100, step=1)
AQE Runtime Decision:
• Detected: Filter reduced left side to 100 rows
• Decision: Switch to BroadcastHashJoin
• Action: Broadcast 100 rows instead of shuffling
Final Plan (after AQE):
BroadcastHashJoin [key#0], [key#5], Inner ← Changed mid-execution!
:- BroadcastExchange
: +- Filter (key#0 < 100)
: +- Range (0, 10000000, step=1)
+- Range (0, 100, step=1)
Performance:
• SortMergeJoin: 12 seconds (expensive shuffle)
• BroadcastHashJoin: 0.8 seconds (15× faster!)
AQE made the query 15× faster automatically!
"""Result: AQE switches join strategies at runtime based on actual data sizes
Memory Management & Garbage Collection
Understanding Spark's memory model prevents OutOfMemoryErrors and optimizes performance. Spark divides executor memory into execution, storage, and overhead regions.
Spark Memory Model (Unified Memory Management): memoryOverhead is a SEPARATE allocation, not carved out of executor.memory - the cluster manager (YARN/Kubernetes) requests executor.memory + memoryOverhead as the container's total size. Container requested from cluster manager (e.g., 16 GB + 1.6 GB overhead) ┌──────────────────────────────────────────────────────────────┐ │ Executor Memory / JVM Heap (16 GB) │ │ ┌────────────────────────────────────────────────────────┐ │ │ │ Reserved Memory (300 MB, fixed) │ │ │ │ Reserved for internal Spark objects │ │ │ ├────────────────────────────────────────────────────────┤ │ │ │ Managed Memory: (heap - 300MB) × memory.fraction │ │ │ │ ≈ 9.4 GB with the 0.6 default │ │ │ │ ┌──────────────────────────────────────────────────┐ │ │ │ │ │ Storage Memory (storageFraction=0.5 → ~4.7 GB) │ │ │ │ │ │ • Cached DataFrames (.cache(), .persist()) │ │ │ │ │ │ • Broadcast variables │ │ │ │ │ │ • Can borrow from execution if idle │ │ │ │ │ ├──────────────────────────────────────────────────┤ │ │ │ │ │ Execution Memory (1 - storageFraction → ~4.7GB) │ │ │ │ │ │ • Shuffles, joins, sorts, aggregations │ │ │ │ │ │ • Can borrow from storage if idle │ │ │ │ │ │ • Spills to disk if full │ │ │ │ │ └──────────────────────────────────────────────────┘ │ │ │ ├────────────────────────────────────────────────────────┤ │ │ │ Unmanaged (heap - 300MB) × (1 - memory.fraction) │ │ │ │ User data structures, UDF working memory, etc. │ │ │ └────────────────────────────────────────────────────────┘ │ ├──────────────────────────────────────────────────────────────┤ │ Overhead Memory (outside the JVM heap, ~10% of executor.mem)│ │ Off-heap: thread stacks, NIO buffers, JVM/native overhead │ └──────────────────────────────────────────────────────────────┘ Key configs: spark.executor.memory = 16g # JVM heap size spark.memory.fraction = 0.6 # Managed memory (default 0.6) spark.memory.storageFraction = 0.5 # Storage vs execution split spark.executor.memoryOverhead = 1.6g # Additional, outside the heap Formula: Managed = (executor.memory - 300MB) × memory.fraction Storage = Managed × storageFraction Execution = Managed × (1 - storageFraction)
Spark divides memory into storage (caching) and execution (shuffles/joins) regions
Optimizing Memory Configuration
# Example: Processing 1 TB of data with 10 executors
# ❌ BAD: Undersized executors (2g is too small) → frequent spills to disk
spark_bad = SparkSession.builder \
.config("spark.executor.memory", "2g") \
.config("spark.executor.cores", "1") \
.getOrCreate()
# Each executor: 2 GB × 0.6 × 0.4 = 480 MB execution memory
# Large join needs 5 GB → spills to disk (100× slower!)
"""
Executor metrics (from Spark UI):
Shuffle Spill (Memory): 0 GB
Shuffle Spill (Disk): 47 GB ← Spilled 47 GB to disk!
Task Time: 45 minutes
"""
# ✅ GOOD: Right-sized executors (16 GB memory, 4 cores, higher usable fraction)
spark_good = SparkSession.builder \
.config("spark.executor.memory", "16g") \
.config("spark.executor.cores", "4") \
.config("spark.executor.memoryOverhead", "2g") \
.config("spark.memory.fraction", "0.75") \
.getOrCreate()
# Each executor: 16 GB × 0.75 × 0.4 = 4.8 GB execution memory
# Large join fits in memory → no disk spills!
"""
Executor metrics:
Shuffle Spill (Memory): 0 GB
Shuffle Spill (Disk): 0 GB ← No spills!
Task Time: 3 minutes ← 15× faster!
"""
# Rule of thumb for executor sizing:
# • executor.cores: 4-5 (diminishing returns beyond 5)
# • executor.memory: 16-64 GB per executor
# • Total cores: executor.cores × num_executors ≤ cluster cores
# • Memory overhead: 10% of executor.memory (auto-calculated)
print("\n💡 Memory Configuration Recommendations:")
print(" Small job (<10 GB): 2 executors × 8 GB = 16 GB")
print(" Medium job (<100 GB): 5 executors × 16 GB = 80 GB")
print(" Large job (>1 TB): 20 executors × 32 GB = 640 GB")Result: Proper memory sizing eliminates disk spills and speeds up jobs 10-100×
Garbage Collection Tuning
# GC overhead problem: Tasks spend 50%+ time in garbage collection!
# ❌ BAD: Old parallel GC causes long pauses
spark_default = SparkSession.builder \
.config("spark.executor.extraJavaOptions",
"-XX:+UseParallelGC") \
.getOrCreate()
"""
GC metrics (from Spark UI → Executors tab):
GC Time: 23 min / 45 min total (51% GC overhead!)
Task deserialization: Very slow
"""
# ✅ GOOD: G1GC with tuned parameters
spark_tuned = SparkSession.builder \
.config("spark.executor.memory", "16g") \
.config("spark.executor.extraJavaOptions",
"-XX:+UseG1GC " + # G1 garbage collector
"-XX:G1HeapRegionSize=32m " + # Larger regions
"-XX:InitiatingHeapOccupancyPercent=35 " + # Trigger GC earlier
"-XX:ConcGCThreads=4 " + # Concurrent GC threads
"-XX:+PrintGCDetails " + # Log GC for debugging
"-XX:+PrintGCTimeStamps") \
.getOrCreate()
"""
GC metrics with G1GC:
GC Time: 2 min / 15 min total (13% GC overhead) ← Much better!
Pause times: < 100ms (vs seconds with ParallelGC)
GC overhead reduced from 51% to 13% → 2× speedup!
"""
# When to tune GC:
# • GC time > 20% of task time (check Spark UI)
# • Frequent "long GC pause" warnings in logs
# • Tasks failing with OutOfMemoryError
# GC Best Practices:
print("\n🗑️ Garbage Collection Best Practices:")
print(" 1. Use G1GC for executors > 4 GB")
print(" 2. Increase heap size before tuning GC")
print(" 3. Monitor GC metrics in Spark UI")
print(" 4. Use off-heap memory for large datasets")
print(" 5. Reduce object creation (use primitive types)")Result: G1GC reduces GC overhead from 51% to 13%, doubling overall speed
Shuffle Optimization: The Biggest Bottleneck
Shuffle is the most expensive operation in Spark, moving data across the network between executors. Operations like groupBy, join, and repartition trigger shuffles. Optimizing shuffles can make queries 10-100× faster.
What happens during a shuffle: Stage 1: Map Side Stage 2: Reduce Side ┌───────────────────┐ ┌───────────────────┐ │ Executor 1 │ │ Executor 1 │ │ ┌───────────────┐ │ │ ┌───────────────┐ │ │ │ Partition 0 │ │───────────────>│ │ Partition 0 │ │ │ │ key=A: 10 MB │ │ Shuffle write │ │ All key=A data│ │ │ │ key=B: 5 MB │ │ (serialize, │ │ from all execs│ │ │ └───────────────┘ │ write to disk) │ └───────────────┘ │ ├───────────────────┤ ├───────────────────┤ │ Executor 2 │ │ Executor 2 │ │ ┌───────────────┐ │ │ ┌───────────────┐ │ │ │ Partition 1 │ │───────────────>│ │ Partition 1 │ │ │ │ key=A: 8 MB │ │ Network │ │ All key=B data│ │ │ │ key=C: 12 MB │ │ transfer │ │ from all execs│ │ │ └───────────────┘ │ (fetch data) │ └───────────────┘ │ └───────────────────┘ └───────────────────┘ Shuffle stages: 1. Map-side: Partition data by key, serialize, write to disk 2. Network: Transfer data between executors 3. Reduce-side: Fetch data, deserialize, merge Cost breakdown: • Disk I/O: 30% (map writes + reduce reads) • Network: 40% (data transfer) • CPU: 30% (serialize/deserialize) Shuffle is expensive! Minimize whenever possible.
Shuffle involves disk writes, network transfer, and deserialization overhead
Minimizing Shuffles with Smart Transformations
# Create large datasets
transactions = spark.range(0, 100_000_000) \
.selectExpr("id as txn_id",
"(id % 1000) as user_id",
"rand() * 1000 as amount")
users = spark.range(0, 1000) \
.selectExpr("id as user_id", "'User' || id as user_name")
# ❌ BAD: Multiple shuffles
# Shuffle 1: repartition, Shuffle 2: groupBy, Shuffle 3: join (if not
# broadcast), Shuffle 4: orderBy
result_bad = transactions \
.repartition(200) \
.groupBy("user_id") \
.agg(_sum("amount")) \
.join(users, "user_id") \
.orderBy("user_id")
# Count shuffles in plan
result_bad.explain()
"""
Physical Plan:
Sort [user_id ASC]
+- Exchange rangepartitioning(user_id, 200) ← Shuffle 4
+- SortMergeJoin [user_id], [user_id]
:- Exchange hashpartitioning(user_id) ← Shuffle 3
: +- HashAggregate [user_id], [sum(amount)]
: +- Exchange hashpartitioning(user_id) ← Shuffle 2
: +- HashAggregate [user_id]
: +- Exchange RoundRobinPartitioning(200) ← Shuffle 1
+- Exchange hashpartitioning(user_id)
Total: 5 shuffles! (100 GB shuffled across network)
"""
# ✅ GOOD: Eliminate unnecessary shuffles
# Shuffle 1 (groupBy) is necessary; the join broadcasts (no extra shuffle),
# and there's no final orderBy() since it isn't needed
result_good = transactions \
.groupBy("user_id") \
.agg(_sum("amount")) \
.join(
users.hint("broadcast"),
"user_id"
)
result_good.explain()
"""
Physical Plan:
BroadcastHashJoin [user_id], [user_id]
:- HashAggregate [user_id], [sum(amount)]
: +- Exchange hashpartitioning(user_id, 200) ← Only 1 shuffle!
: +- HashAggregate [user_id]
+- BroadcastExchange ← Broadcast (no shuffle)
Total: 1 shuffle (10 GB shuffled vs 100 GB)
Speedup: 10× faster!
"""
# Shuffle reduction techniques:
print("\n🔀 Shuffle Optimization Techniques:")
print(" 1. Use broadcast joins for small tables (< 100 MB)")
print(" 2. Filter BEFORE shuffle operations")
print(" 3. Avoid unnecessary repartition()")
print(" 4. Use coalesce() instead of repartition() to reduce partitions")
print(" 5. Persist intermediate results to avoid re-shuffling")Result: Reducing shuffles from 5 to 1 provides 10× speedup
Shuffle Partitioning and Tuning
# Default shuffle partitions: 200 (often too many or too few!)
# ❌ BAD: 200 partitions for small data
small_data = spark.range(1000).groupBy("id").count()
small_data.explain()
"""
Exchange hashpartitioning(id, 200) ← 200 partitions for 1000 rows!
Each partition: ~5 rows (wasted overhead)
"""
# ✅ GOOD: Tune partitions based on data size
# Rule of thumb: 128 MB - 200 MB per partition
# For small data (< 1 GB): Use fewer partitions
spark.conf.set("spark.sql.shuffle.partitions", "10")
# For medium data (1-100 GB): Default 200 is OK
spark.conf.set("spark.sql.shuffle.partitions", "200")
# For large data (> 1 TB): Use more partitions
spark.conf.set("spark.sql.shuffle.partitions", "2000")
# Or let AQE decide automatically!
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.advisoryPartitionSizeInBytes", "128MB")
"""
AQE will automatically:
• Start with 200 partitions
• Measure actual partition sizes after shuffle
• Coalesce small partitions to target 128 MB each
• Result: Optimal partition count without manual tuning!
"""
# Monitoring shuffle performance
def analyze_shuffle(df):
"""Analyze shuffle characteristics"""
# Force execution
df.cache().count()
# Check Spark UI metrics
print("\n📊 Shuffle Metrics (from Spark UI):")
print(" • Shuffle Read: Check 'Input Size / Records'")
print(" • Shuffle Write: Check 'Shuffle Write Size / Records'")
print(" • Spill (Memory): Should be 0 (data fits in memory)")
print(" • Spill (Disk): Should be 0 (no disk spills)")
print(" • Task Duration: Should be similar across tasks (no skew)")
# Example: Optimize shuffle for 10 GB dataset
# 10 GB / 128 MB per partition = ~80 partitions
spark.conf.set("spark.sql.shuffle.partitions", "80")
large_result = transactions.groupBy("user_id").agg(_sum("amount"))
analyze_shuffle(large_result)
"""
Before tuning (200 partitions):
Partition size: ~50 MB (too small, high overhead)
Task time: 5 seconds per task
Total: 200 tasks × 5 sec = 16 minutes
After tuning (80 partitions):
Partition size: ~125 MB (optimal)
Task time: 8 seconds per task
Total: 80 tasks × 8 sec = 10 minutes
37% faster with better partitioning!
"""Result: Right-sized partitions (128 MB each) reduce overhead and improve speed by 37%
Production Spark Tuning Checklist
| Category | Optimization | Configuration / Code | Impact |
|---|---|---|---|
| Query Planning | Enable CBO with statistics | ANALYZE TABLE, spark.sql.cbo.enabled=true | 2-5× faster joins |
| Execution | Enable Adaptive Query Execution | spark.sql.adaptive.enabled=true | 2-10× speedup |
| Memory | Right-size executors | executor.memory=16-64g, executor.cores=4-5 | Eliminate spills |
| GC | Use G1GC for large heaps | -XX:+UseG1GC | 50% less GC time |
| Shuffle | Broadcast small tables | df.hint("broadcast") | 10× faster joins |
| Shuffle | Tune partition count | sql.shuffle.partitions=data_size_GB/0.128 | 30-50% faster |
| I/O | Use columnar formats | Parquet, ORC | 10-100× less data read |
| I/O | Partition data by common filters | partitionBy("date", "country") | 10-100× faster queries |
| Caching | Cache hot DataFrames | df.cache() or df.persist(MEMORY_AND_DISK) | Skip recomputation |
| Code | Filter early | Apply filters before joins/groupBy | Process less data |
Key Takeaways
- Catalyst optimizer: Rewrites queries with predicate pushdown, column pruning
- Tungsten engine: Whole-stage codegen for 5-10× speedup
- AQE: Runtime optimization with dynamic join switching, skew handling
- Memory model: Storage (cache) vs execution (shuffle) regions
- GC tuning: G1GC reduces overhead from 51% to 13%
- Shuffle optimization: Broadcast joins, filter early, right-size partitions
- Partition sizing: Target 128-200 MB per partition
- CBO: Compute statistics for optimal join strategies
Remember: Production Spark performance requires understanding the internals. Catalyst automatically optimizes queries but needs statistics (ANALYZE TABLE) to make smart decisions. Tungsten's whole-stage code generation eliminates 90% of virtual function calls, keep it enabled! AQE (Spark 3.0+) fixes bad plans at runtime and should always be enabled. Memory sizing prevents OutOfMemoryErrors: use 4-5 cores and 16-64 GB per executor. G1GC reduces garbage collection overhead for heaps > 4 GB. Shuffle is the biggest bottleneck, broadcast small tables (< 100 MB), filter before groupBy/join, and tune spark.sql.shuffle.partitions based on data size. Companies like Netflix, Apple, and Uber process petabytes daily by applying these optimizations. A well-tuned Spark job can be 100× faster than naive code!