SPARK Reference Guide

Revision Time: 10 mins

Optimization & Performance Tuning Reference

Essential Spark optimization strategies: Broadcast Joins, AQE, Salting, Caching, and Shuffle Partitions.

broadcast (Broadcast Join)

Return: DataFrame

Broadcasts a small DataFrame to all executor nodes to avoid costly network shuffle during joins.

Used In: Lookup Joins, Dimension-Fact Joins

Syntax signature:

from pyspark.sql.functions import broadcast

large_df.join(broadcast(small_df), 'user_id')

Code snippet:

python

orders_df.join(broadcast(country_lookup_df), 'country_code')

Expected Output:Fast BroadcastHashJoin without shuffling large_df

Common Mistakes:Broadcasting large tables (>1GB), causing Driver OutOfMemory errors during broadcast payload collection.

Remember: Default auto-broadcast threshold is 10MB (`spark.sql.autoBroadcastJoinThreshold = 10485760`). Manually wrapping small lookup tables up to 100MB with `broadcast()` delivers 10x-50x speedups.

Adaptive Query Execution (AQE)

Return: Configuration

Re-optimizes physical query execution plans at runtime based on actual stage metrics.

Used In: Production Jobs, Dynamic Partition Coalescing, Skew Mitigation

Syntax signature:spark.conf.set('spark.sql.adaptive.enabled', 'true')

Code snippet:

python

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')

Expected Output:Dynamic runtime query optimization enabled

Remember: AQE is enabled by default in Spark 3.0+. It dynamically coalesces small shuffle partitions, converts sort-merge joins to broadcast joins at runtime, and handles data skew automatically.

Salting (Data Skew Mitigation)

Return: DataFrame

Solves data skew where 1 key (e.g., NULL or 'UNKNOWN') receives 90% of data, overloading a single reducer partition.

Used In: Handling Skewed Joins, Straggler Mitigation

Syntax signature:

# Add random salt key to skewed dataframe:
df = df.withColumn('salt', (rand() * 10).cast('int'))
# Join on key AND salt

Code snippet:

python

skewed_df.withColumn('salted_key', concat(col('user_id'), lit('_'), floor(rand() * 10)))
lookup_df_exploded = lookup_df.withColumn('salt', explode(array([lit(i) for i in range(10)])))

Expected Output:Uniformly distributed partitions without straggler tasks

Remember: If 1 executor task runs for 3 hours while all other 99 tasks finish in 2 minutes, you have DATA SKEW. Salting spreads skewed key instances across multiple salt partitions.

spark.sql.shuffle.partitions

Return: Configuration

Controls the default number of partitions used when shuffling data for joins or aggregations.

Used In: Cluster Tuning, Memory Allocation

Syntax signature:spark.conf.set('spark.sql.shuffle.partitions', '200')

Code snippet:

python

# For small datasets (100MB):
spark.conf.set('spark.sql.shuffle.partitions', '10')

# For large datasets (1TB):
spark.conf.set('spark.sql.shuffle.partitions', '2000')

Expected Output:Adjusted shuffle partition count

Common Mistakes:Leaving default 200 partitions on a 50MB dataset (creates 200 micro tasks) or 5TB dataset (causes disk spill OOM).

Remember: Default value is 200. For small datasets (<1GB), 200 partitions creates tiny file overhead. Rule of thumb: Aim for ~100MB - 200MB per shuffle partition.

Dynamic Partition Pruning (DPP)

Return: Configuration

Skips reading irrelevant partition directories at runtime when joining a large partitioned Fact table with a filtered Dimension table.

Used In: Data Warehouse Joins, Star Schema Queries

Syntax signature:spark.conf.set('spark.sql.optimizer.dynamicPartitionPruning.enabled', 'true')

Code snippet:

python

# DPP activates automatically when joining on partition keys:
fact_sales.join(dim_date.filter(col('year') == 2024), 'date_id')

Expected Output:

date_id	sales	year
20241	15000.0	2024
20242	22000.0	2024

Common Mistakes:Joining on a non-partition column or using non-equi join conditions where DPP cannot apply.

Remember: DPP kicks in automatically in Spark 3.0+ when joining a partitioned table with a filtered table on the partition column, reducing I/O drastically.

Bucketing (bucketBy)

Return: None

Pre-sorts and pre-partitions tables on disk by a specific key to eliminate network shuffles during join operations.

Used In: Repeated Joins on Large Tables, Data Mart Optimization

Syntax signature:df.write.bucketBy(numBuckets, colName).sortBy(sortCol).saveAsTable(tableName)

Code snippet:

python

df1.write.bucketBy(16, 'user_id').sortBy('user_id').saveAsTable('bucketed_users')
df2.write.bucketBy(16, 'user_id').sortBy('user_id').saveAsTable('bucketed_orders')

# Join operates with ZERO network shuffle:
spark.table('bucketed_users').join(spark.table('bucketed_orders'), 'user_id')

Expected Output:

user_id	name	order_id
1001	Alice	ORD_99

Common Mistakes:Using `df.write.parquet()` instead of `df.write.saveAsTable()`. Bucketing metadata is stored in Metastore tables, not raw parquet files.

Remember: Both tables MUST have the exact same number of buckets (`numBuckets`) and identical bucket key columns for Spark to skip the shuffle.

Memory Fractions (Storage vs Execution)

Return: Configuration

Configures executor JVM memory allocation split between Execution (joins/shuffles) and Storage (caches/broadcasts).

Used In: Executor Tuning, Out-of-Memory Debugging

Syntax signature:

spark.memory.fraction = 0.6  # Total Spark RAM fraction
spark.memory.storageFraction = 0.5  # Storage fraction of Spark RAM

Code snippet:

python

# Submit job tuned for heavy shuffles:
--conf spark.memory.fraction=0.8 \
--conf spark.memory.storageFraction=0.2

Expected Output:

Memory Configuration Applied:
- spark.memory.fraction: 0.8
- spark.memory.storageFraction: 0.2

Remember: Spark 1.6+ Unified Memory Manager dynamically borrows memory between Execution and Storage. If jobs crash with `FetchFailedException`, increase `spark.memory.fraction`.

Kryo Serialization (spark.serializer)

Return: Configuration

Replaces Java's slow default object serializer with Kryo serialization for 2x-10x faster serialization speed and smaller memory footprints.

Used In: High-throughput Shuffles, RDD Caching

Syntax signature:spark.serializer = org.apache.spark.serializer.KryoSerializer

Code snippet:

python

conf = SparkConf()\
    .set('spark.serializer', 'org.apache.spark.serializer.KryoSerializer')\
    .set('spark.kryoserializer.buffer.max', '1024m')
spark = SparkSession.builder.config(conf=conf).getOrCreate()

Expected Output:Serializer Enabled: org.apache.spark.serializer.KryoSerializer

Common Mistakes:Forgetting to increase `spark.kryoserializer.buffer.max` when handling large nested objects.

Remember: Essential when caching custom RDD objects or shuffling large objects across network sockets.

Off-Heap Memory Allocation

Return: Configuration

Allocates memory outside the Java Virtual Machine (JVM) heap to avoid Java Garbage Collection (GC) pauses on huge datasets.

Used In: Ultra Large Scale Clusters, Garbage Collection Mitigation

Syntax signature:

spark.memory.offHeap.enabled = true
spark.memory.offHeap.size = 10g

Code snippet:

python

--conf spark.memory.offHeap.enabled=true \
--conf spark.memory.offHeap.size=8g

Expected Output:Off-Heap Memory Allocator Active (8GB)

Remember: Prevents long JVM GC pause freezes on 50GB+ executor RAM nodes.

Executor & Core Sizing Rules

Return: Cluster Config

Calculates optimal executor memory and core configurations to avoid HDFS I/O bottlenecks and GC overhead.

Used In: Production YARN / Kubernetes Spark Submit Tuning

Syntax signature:--executor-cores 5 --executor-memory 19g --num-executors 50

Code snippet:

python

# Optimal Rule of Thumb:
# 5 Cores per Executor (Max HDFS throughput)
# ~4GB RAM per Core -> 20GB Executor Memory + 10% Overhead

Expected Output:

Cluster Allocation:
- Cores per executor: 5
- Memory per executor: 19GB

Common Mistakes:Setting `--executor-cores 16` causing 5-minute Garbage Collection pauses.

Remember: NEVER allocate 16+ cores to 1 executor (causes JVM GC contention). NEVER allocate 1 core to 1 executor (loses multithreading benefits). 4 to 5 cores per executor is optimal.

Previous TopicWindow Functions Next TopicSpark Executor Memory Architecture