Caching and persistence: real-world lessons and examples

Caching and persistence: real-world lessons and examples

In production Spark workloads, data often flows through multiple transformations and actions. Without strategic caching and persistence, you'll recompute the same DataFrame dozens of times, burning cluster resources and wall-clock time. This guide walks through caching and persistence patterns we've refined across e-commerce pipelines, real-time analytics, and batch ETL jobs.

Why caching matters in production

Spark's lazy evaluation model is powerful but deceptive. When you chain ten transformations together, nothing actually happens until you call an action like collect(), write(), or count(). The moment you trigger that first action, Spark rebuilds the entire DAG from scratch. If you then run a second action on the same DataFrame, Spark repeats all ten transformations again.

In real-world scenarios, this becomes catastrophic. We once processed a customer event log through five stages of aggregation, then needed both a summary metric and detailed rollup by region. Without caching, we ran the entire pipeline twice. The job went from 6 minutes to 18 minutes—purely from redundant computation.

Caching and persistence are your tools to break this cycle. By materializing data in memory or disk, you reuse the result across multiple actions without recomputation.

cache() for repeated use

cache() is the simplest, most common persistence method. It stores a DataFrame in memory across the cluster and triggers lazy evaluation on the first action.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("CachingDemo").getOrCreate()

# Read raw user events
events_df = spark.read.parquet("s3://data-lake/events/")

# Apply transformations
cleaned = events_df.filter(events_df.event_type != "bot")
aggregated = cleaned.groupBy("user_id").count()

# Cache before multiple actions
aggregated.cache()

# First action: triggers computation and caches result
summary_count = aggregated.count()
print(f"Total events: {summary_count}")

# Second action: reads from cache, no recomputation
top_users = aggregated.filter("count > 100").collect()
print(f"Power users: {len(top_users)}")

# Third action: also uses cache
aggregated.write.mode("overwrite").parquet("s3://outputs/top_users/")

The first call to count() forces Spark to execute the filter and groupBy, then stores the result in memory. The subsequent filter().collect() and write()` operations skip recomputation entirely.

In a trading analytics platform we worked on, we cached hourly price summaries before computing percentiles, volatility, and correlation matrices. Caching cut execution time by 65% because the hourly aggregation was expensive and reused across six downstream analyses.

persist() with storage levels

persist() is more flexible than cache(). It lets you specify where and how data is stored: memory, disk, or both. This is critical when data volume exceeds available memory.

cache() is actually shorthand for persist(StorageLevel.MEMORY_AND_DISK).

from pyspark import StorageLevel

# Scenario: processing a 50GB clickstream log
clickstream = spark.read.parquet("s3://data-lake/clickstream/")

# Heavy transformation: joining with product metadata
enriched = clickstream.join(
    spark.read.parquet("s3://data-lake/products/"),
    on="product_id",
    how="left"
)

# Store in serialized format to save memory
enriched.persist(StorageLevel.MEMORY_ONLY_SER)

# Multiple analyses
revenue_by_category = enriched.groupBy("category").sum("price").collect()
sessions_by_device = enriched.groupBy("device_type").count().collect()
bounce_rate = enriched.filter("time_on_page < 5").count()

In a SaaS analytics engine, we discovered that serialized storage reduced memory usage by 40% while keeping query latency under 2 seconds. Uncompressed DataFrames spilled to disk constantly, causing GC pauses and timeouts.

DISK_ONLY is rare but useful for checkpoints in long pipelines. If a node fails mid-workflow, you can resume from a persisted checkpoint without rerunning expensive transformations.

from pyspark import StorageLevel

# Heavy computation that might fail mid-pipeline
stage_one = spark.read.parquet("raw_data/")
stage_one = stage_one.repartition(200)
stage_one = stage_one.filter("quality_score > 0.8")

# Persist to disk as a checkpoint
stage_one.persist(StorageLevel.DISK_ONLY)
stage_one.count()  # Force evaluation

# If downstream fails, stage_one is still available
stage_two = stage_one.join(dimension_table, on="id", how="inner")
stage_two.write.mode("overwrite").parquet("output/")

Unpersist when done

Persisted data occupies cluster resources until the Spark session ends or you explicitly call unpersist(). In long-running workflows or shared cluster environments, leaked cache blocks cause memory pressure and evict other workloads' data.

from pyspark import StorageLevel

user_events = spark.read.parquet("events/")
cleaned = user_events.filter("user_id IS NOT NULL")
cleaned.persist(StorageLevel.MEMORY_AND_DISK)

# Run analyses
print(cleaned.count())
print(cleaned.groupBy("event_type").count().collect())

# Done with cleaned data
cleaned.unpersist()
print("Cache freed")

In a ride-hailing platform, we cached hourly demand predictions across 50+ regional sub-tasks. A junior engineer forgot to unpersist after each region's analysis. By hour 24, the driver nodes were swapping to disk, and job completion time doubled. A single unpersist() call fixed it.

When working with temporary tables or broadcast joins, clean up explicitly:

broadcast_df = spark.read.parquet("reference_data/")
broadcast_df.persist(StorageLevel.MEMORY_ONLY)

# Use in multiple joins
result_1 = fact_table_1.join(broadcast_df, on="key")
result_2 = fact_table_2.join(broadcast_df, on="key")

# Unpersist to free memory for downstream pipelines
broadcast_df.unpersist()

# Continue with result_1, result_2 without the broadcast overhead

Do not cache everything

Caching is powerful but not free. Memory is finite. Caching the wrong data wastes resources and can cause out-of-memory errors. Use these heuristics:

• Cache only if reused: If a DataFrame is used once and discarded, caching adds overhead. Only cache DataFrames referenced in two or more actions.
• Cache after expensive operations: Cache after groupBy, join, or complex filtering. Don't cache raw, unprocessed data that's cheap to recompute.
• Monitor cache hit rates: Use Spark UI to check if your cached data is actually being used. If a persisted DataFrame is evicted or rarely accessed, reconsider.
• Be aware of dataset size: If a DataFrame is 80% of available executor memory, caching it leaves little room for processing. Use MEMORY_ONLY_SER or MEMORY_AND_DISK instead.

from pyspark import StorageLevel

# Bad: caching raw, unprocessed data used once
raw_logs = spark.read.parquet("logs/").cache()
summary = raw_logs.groupBy("status").count()
# raw_logs is never reused; cache is wasted

# Good: cache after aggregation, before multiple uses
raw_logs = spark.read.parquet("logs/")
summarized = raw_logs.groupBy("status").count()
summarized.cache()

# Reuse multiple times
error_logs = summarized.filter("status >= 400")
success_rate = summarized.filter("status < 400").collect()
error_logs.write.mode("overwrite").parquet("errors/")

Common mistakes

• Forgetting to trigger evaluation: Calling cache() alone does nothing. You must call an action like count() or write() to force evaluation and actually cache the data.
• Caching inside loops: If you cache a DataFrame inside a loop, each iteration caches a new copy. Move cache outside the loop.
• Mixing cache and non-cache in transformations: If you cache DataFrame A, then transform it without caching (B = A.filter(...)), subsequent uses of B recompute from A. Cache B if you plan to reuse it.
• Over-relying on MEMORY_ONLY: If data doesn't fit in memory, Spark discards it, and the next action recomputes. Use MEMORY_AND_DISK or MEMORY_ONLY_SER for safety.
• Not unpersisting in interactive shells: In Jupyter or Databricks notebooks, cached data persists across cells. Unpersist explicitly or restart the kernel.

Key takeaways

• cache() is persist(MEMORY_AND_DISK)—use it for DataFrames reused in multiple actions.
• Choose storage levels deliberately: MEMORY_ONLY_SER for large data, MEMORY_ONLY for small reference tables, DISK_ONLY for checkpoints.
• Always call an action after cache() or persist() to trigger materialization.
• Call unpersist() when done to free cluster memory and prevent resource contention.
• Cache only after expensive operations and only if the DataFrame is reused. Monitor the Spark UI to verify cache effectiveness.
• In production, test cache behavior under realistic data volumes. What works locally may cause OOM errors at scale.