Spark Tuning - Performance Optimizations: Theoretical Quiz

This assessment details the runtime mechanics of Adaptive Query Execution (AQE) and cache memory management.

Scenario 1: Adaptive Query Execution (AQE) Optimization Modes

The Scenario

An enterprise data pipeline runs on a cluster with AQE enabled. The log logs detail execution plans restructuring on-the-fly:

INFO AQE: Coalescing 200 shuffle partitions to 8 partitions
INFO AQE: Switching SortMergeJoin to BroadcastHashJoin

The Questions

1. Detail the three core runtime optimizations implemented by Adaptive Query Execution (AQE).
2. How does AQE acquire the partition statistics required to make decisions mid-query?

Detailed Solution & Architectural Analysis

1. The Three AQE Optimizations

1. Dynamically Coalescing Shuffle Partitions: By default, Spark uses 200 shuffle partitions (spark.sql.shuffle.partitions). If the output is small, this results in hundreds of tiny files. AQE monitors stage outputs, aggregates tiny partitions, and coalesces them into larger, uniform partitions (e.g. 200 down to 8), reducing metadata bloat and disk read latency.

2. Dynamically Switching Join Strategies: If a query joins two large tables, Spark plans a Sort-Merge Join. If a filter reduces one table's actual output size below the broadcast threshold (e.g., to 5MB), AQE dynamically switches the physical plan to a Broadcast Hash Join, eliminating network shuffles.

3. Dynamically Handling Skew Joins: If AQE detects partition size skew (e.g. Partition 5 is 20x larger than others), it splits that heavy partition into smaller sub-partitions, reads the matching join side, and joins them in parallel, preventing single-executor straggler delays.

2. Stats Collection Mechanics

AQE divides the execution plan into Query Stages bounded by shuffles. Before beginning a stage, it completes the previous shuffle stage. The executors write shuffle partition size metrics to disk. AQE parses these size records, compiles the partition statistics, and optimizes the downstream logical plans on-the-fly.

Scenario 2: Memory Tuning Cache Heap fractions

The Scenario

A cluster administrator configures spark.memory.fraction and spark.memory.storageFraction. When users run cache-heavy pipelines, data processing tasks spill to disk.

The Questions

1. Map the allocation of JVM Heap space between Storage Memory, Execution Memory, and User Memory.
2. How do Storage and Execution memory dynamically borrow space from each other?

Detailed Solution & Architectural Analysis

1. JVM Memory Layout Fractions

Spark manages executor memory allocations using strict heap boundaries:

• Execution Memory: Dedicated to storing temporary data required by tasks (shuffle buffers, hash aggregation maps, sort buffers).
• Storage Memory: Dedicated to caching DataFrames (.cache(), .persist()) and broadcast variables.
• User Memory: Reserved for user-defined JVM structures, metadata, and Spark internal classes.

2. Dynamic Memory Borrowing Rules

Since Spark 1.6, execution and storage memory co-exist inside a shared pool:

• Execution takes priority: If execution memory requires space (due to large shuffles), it can evict cached storage partitions out of memory (dropping them to disk or out of cache) to claim the RAM.
• Storage cannot evict execution: Storage can borrow free space from execution memory if it is idle. However, if execution suddenly requires that space, Spark forces storage to evict its blocks, ensuring tasks never crash due to cache allocation.

Scenario 3: Garbage Collection Optimization under cache loads

The Scenario

A developer caches a 150GB dataset using MEMORY_ONLY. The job slows down, spending 45% of its time running JVM Garbage Collection, leading to executor timeouts.

The Questions

1. Explain how storing raw deserialized JVM objects on the heap triggers GC delays.
2. Provide two caching optimizations to reduce heap memory pressure.

Detailed Solution & Architectural Analysis

1. Caching Object Bloat & GC pressure

• The Issue: Under MEMORY_ONLY, cached records are stored as millions of independent, deserialized Java objects on the JVM heap.
• GC Scan overhead: When the JVM performs a garbage collection pass (like G1GC), it must scan the headers of every single active object on the heap to see if it is eligible for collection. Scanning millions of persistent cache objects takes a long time, freezing all execution threads (GC pauses).

2. Two Caching Optimizations

1. Use MEMORY_ONLY_SER: Serializes the cached objects into a single, compact byte-array block per partition. The JVM sees only 1 byte-array object per partition instead of millions of small records, reducing GC scanning overhead to near-zero.
2. Utilize Off-Heap Caching (MEMORY_ONLY_SER_2 / OFF_HEAP): Configures Spark to store cached blocks in off-heap memory outside the control of the JVM GC engine.

Scenario 4: Caching vs. Persistence Execution Semantics

The Scenario

A junior developer calls .cache() and expects Spark to populate memory instantly. They notice that memory remains unallocated in the Spark UI until a downstream action runs.

The Questions

1. Why is .cache() a lazy transformation rather than an instant memory allocation?
2. How can you force immediate caching in Spark?

Detailed Solution & Architectural Analysis

1. Caching is Lazy

• .cache() contract: In Spark, .cache() is simply a metadata declaration. It updates the RDD/DataFrame lineage graph to note that once this block is computed, its partitions should be saved in the Storage pool.
• Execution: No memory is allocated because no physical tasks have run to load the raw file bytes or perform computations.

2. Forcing Cache Allocation

To trigger cache allocation instantly, call an Action immediately after the cache declaration:

df.cache()
df.count() # Action forces Spark to execute the DAG and cache partitions in memory