Caching & Persistence

In large-scale ETL pipelines, you often need to reuse an intermediate DataFrame multiple times (e.g. performing multiple aggregates, running iterative machine learning, or creating different branch tables). By default, Spark's lazy evaluation will re-compute that entire DataFrame from scratch every single time an action is triggered.

To avoid this massive re-computation penalty, you can cache the computed rows directly in memory.

Caching vs. Persistence

Spark provides two methods to store data in memory:

1. cache(): A quick shortcut that uses the default storage level:
   • For RDDs: MEMORY_ONLY (unserialized in RAM).
   • For DataFrames: MEMORY_AND_DISK (unserialized in RAM, spills to disk when memory overflows).
2. persist(storageLevel): The complete, professional API that allows you to specify a custom Storage Level depending on your memory limits.

Storage Levels Reference

Import pyspark.StorageLevel to configure persistence:

PySpark Code Example: Caching & Eviction

Here is a complete script demonstrating DataFrame persistence, execution timing comparisons, and manual memory freeing (unpersist):

import time
from pyspark.sql import SparkSession
from pyspark import StorageLevel

# 1. Setup Spark
spark = SparkSession.builder \
    .appName("Caching and Persistence") \
    .master("local[*]") \
    .getOrCreate()

# 2. Load and create a large dummy dataset
# We create 10,000,000 integers to observe memory speeds
df = spark.range(1, 10000000).withColumn("value", F.col("id") * 2)

# 3. Persist DataFrame in memory using Serialization (saves memory space)
df.persist(StorageLevel.MEMORY_AND_DISK_SER)

# 4. Trigger Action A (This builds the cache - takes a few seconds)
start_time = time.time()
count_a = df.count()
print(f"Action A (Cache build) - Count: {count_a}, Time taken: {time.time() - start_time:.2f} seconds")

# 5. Trigger Action B (Reads directly from memory cache - extremely fast!)
start_time = time.time()
count_b = df.count()
print(f"Action B (From Cache) - Count: {count_b}, Time taken: {time.time() - start_time:.2f} seconds")

# 6. Release memory manually (Production Best Practice!)
# This prevents memory leaks and ensures subsequent steps have enough heap space
df.unpersist()

Production Best Practices & Pitfalls

• Do Not Cache Everything: Caching consumes executor RAM. If you cache too many DataFrames, Spark will run out of memory, causing executor GC pauses, Out Of Memory (OOM) errors, or excessive disk spelling.
• Always call unpersist(): As soon as you are finished with the cached DataFrame, call unpersist() to free up executor heap space.