Spark DataFrame caching pitfalls: lessons from production

We've all felt the sting of a Spark job that runs smoothly in development but crawls to a halt in production. One of the most deceptive culprits is caching—a feature so appealing in its simplicity that teams often misuse it catastrophically. Spark DataFrame caching pitfalls: lessons from production emerge when we treat caching as a silver bullet rather than a surgical tool, and this article walks through the real-world mistakes we've made and how to avoid them.

Understanding Spark caching: the false promise

Caching a DataFrame in Spark feels like a free lunch. You call .cache() or .persist(), and suddenly subsequent actions on that DataFrame should be faster because the data lives in memory. The problem is that caching consumes memory from a finite pool—your executor's working memory.

In a production Kafka-to-analytics pipeline we managed, a team cached an intermediate DataFrame that held 18 GB of compressed data across a 20-node cluster with 8 GB executor memory per node. They expected 144 GB of total cache capacity, but they didn't account for:

• Spark's own overhead and shuffle buffers
• Data expansion when deserialized
• Competing jobs running on the same cluster
• Garbage collection pressure from repeated eviction

The result was constant cache thrashing: as new tasks needed memory, the eviction policy kicked in, removing previously cached blocks. Each subsequent action triggered a re-read and re-cache cycle, turning their optimization into a performance anchor.

Caching huge data evicts other data

Spark's block manager uses an eviction policy (LRU by default) to manage memory when the cache fills up. When you cache a massive DataFrame, you're not just competing with your own pipeline; you're competing with every other job on the cluster. On a shared cluster, caching a 15 GB DataFrame might evict 15 GB of previously cached data from other jobs or even other stages of your own job.

Consider a multi-stage ETL job we wrote for an e-commerce platform:

df_orders = spark.read.parquet("s3://bucket/orders/")
df_customers = spark.read.parquet("s3://bucket/customers/")

# Stage 1: Join orders and customers
df_joined = df_orders.join(df_customers, on="customer_id")
df_joined.cache()  # 12 GB cached

# Stage 2: Aggregate by region
df_regional = df_joined.groupBy("region").agg({"amount": "sum"})
df_regional.cache()  # 3 GB cached

# Stage 3: Rank regions
df_ranked = df_regional.withColumn("rank", 
    F.row_number().over(Window.orderBy(F.desc("amount"))))
df_ranked.show()

When we added the second cache for df_regional, Spark had to evict parts of df_joined from memory to make room. The third stage then re-materialized df_joined on demand, defeating the entire purpose of the first cache. We learned to cache selectively—only the most expensive intermediate data, only when we're certain it will be reused multiple times without eviction.

Cache after expensive steps only

The strategic placement of cache calls is everything. A good rule: cache after expensive operations like joins, aggregations, or window functions that you'll definitely reuse. Don't cache the raw input data unless you're rereading it multiple times in the same job—usually you're not.

Here's a production pattern from a trading system we built:

# Bad: caching raw data
df_trades = spark.read.parquet("s3://bucket/trades/")
df_trades.cache()  # Unnecessary; reading from parquet is already fast

# Good: cache after expensive join
df_trades = spark.read.parquet("s3://bucket/trades/")
df_market_data = spark.read.parquet("s3://bucket/market_data/")

df_enriched = df_trades.join(
    df_market_data,
    on=["symbol", "timestamp"],
    how="left"
).cache()  # Cache after the join; this is expensive

# Now we can compute multiple metrics from df_enriched
df_metric_1 = df_enriched.groupBy("symbol").agg(...)
df_metric_2 = df_enriched.groupBy("sector").agg(...)

df_metric_1.write.parquet("s3://bucket/metric1/")
df_metric_2.write.parquet("s3://bucket/metric2/")

The join is expensive (O(n log n) in Spark's default sort-merge join). By caching df_enriched, both downstream aggregations reuse the computed join result. If we'd cached the raw inputs, we'd gain almost nothing—Spark's cost-based optimizer already avoids re-reading the same file in a single stage.

Count to materialize when needed

Caching in Spark is lazy. Calling .cache() doesn't immediately load data into memory; it just marks the DataFrame for caching. The actual caching happens when an action triggers evaluation. This is where timing matters.

In a CRM system we supported, a team wrote:

df = spark.read.parquet("path/to/data/")
df_filtered = df.filter(F.col("status") == "active").cache()

# Developer assumed df_filtered was cached here, but it's not

result1 = df_filtered.groupBy("region").count().collect()
# Action 1: df_filtered is materialized and cached here

result2 = df_filtered.groupBy("company_size").count().collect()
# Action 2: uses cached data from Action 1, which is good

The problem arose when another developer added a filter in between:

df = spark.read.parquet("path/to/data/")
df_filtered = df.filter(F.col("status") == "active").cache()

result1 = df_filtered.groupBy("region").count().collect()

# Someone added a new filter expecting lazy evaluation
df_filtered_strict = df_filtered.filter(F.col("created_date") > "2024-01-01")

result2 = df_filtered_strict.groupBy("company_size").count().collect()
# This doesn't use the cache for df_filtered; it creates a new plan

To make caching explicit and intentional, trigger materialization immediately after the cache call:

df_filtered = df.filter(F.col("status") == "active").cache()
df_filtered.count()  # Trigger materialization immediately

# Now subsequent actions definitely use cached data
result1 = df_filtered.groupBy("region").count().collect()
result2 = df_filtered.groupBy("company_size").count().collect()

Calling count() forces Spark to compute and cache the entire DataFrame right away. You know exactly when the cache is populated, and you avoid surprises later.

Unpersist after use

Memory held by cached DataFrames persists until the session ends or you explicitly remove it. In long-running applications—batch jobs that process multiple tables sequentially, or Spark SQL servers handling concurrent queries—failure to unpersist old data starves new operations.

A ride-hailing analytics job we built processed 50 daily tables. Each table required a join with a large dimension table, then multiple aggregations:

tables = ["table_1", "table_2", ..., "table_50"]

for table_name in tables:
    df_fact = spark.read.parquet(f"s3://bucket/{table_name}/")
    df_dim = spark.read.parquet("s3://bucket/dimension/")
    
    df_joined = df_fact.join(df_dim, on="id").cache()
    df_joined.count()  # Materialize
    
    result1 = df_joined.groupBy("city").agg(...)
    result1.write.parquet(f"s3://bucket/output/{table_name}_by_city/")
    
    result2 = df_joined.groupBy("trip_type").agg(...)
    result2.write.parquet(f"s3://bucket/output/{table_name}_by_type/")
    
    # Bug: df_joined is still in cache after the loop iteration

By iteration 30, memory was exhausted, and the job crashed. The fix was simple but critical:

for table_name in tables:
    df_fact = spark.read.parquet(f"s3://bucket/{table_name}/")
    df_dim = spark.read.parquet("s3://bucket/dimension/")
    
    df_joined = df_fact.join(df_dim, on="id").cache()
    df_joined.count()
    
    result1 = df_joined.groupBy("city").agg(...)
    result1.write.parquet(f"s3://bucket/output/{table_name}_by_city/")
    
    result2 = df_joined.groupBy("trip_type").agg(...)
    result2.write.parquet(f"s3://bucket/output/{table_name}_by_type/")
    
    df_joined.unpersist()  # Explicitly free memory

Calling unpersist() removes the DataFrame from cache immediately, freeing memory for the next iteration. In a Spark SQL context or interactive Jupyter notebook, consider using Spark's built-in sql("CLEAR CACHE") after major workloads, though this is more blunt.

Common mistakes

Mistake 1: Caching without measuring. Teams often cache because "it feels right" without profiling to check if the operation is actually expensive or if the data is truly reused. Use Spark's UI to identify bottlenecks before adding cache.

Mistake 2: Ignoring serialization format. By default, cache() uses Java serialization, which is slow and memory-intensive. Use persist(StorageLevel.MEMORY_AND_DISK_SER) or Kryo serialization for large DataFrames to reduce memory footprint:

spark.conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
df.persist(StorageLevel.MEMORY_AND_DISK_SER)

Mistake 3: Mixing cache with shuffle operations carelessly. Caching before a shuffle can consume enormous memory. Shuffle data is temporary; caching it often wastes resources. Cache after the shuffle if you need to reuse aggregated results.

Mistake 4: Assuming cache survives across separate actions in multi-threaded code. If multiple threads execute actions on the same cached DataFrame concurrently, eviction can become unpredictable. Prefer explicit count() calls to ensure materialization before concurrent access.

Key takeaways

• Cache is not free. It consumes executor memory and competes with other data on shared clusters. Treat it as a premium resource.
• Cache strategically. Place cache calls after expensive operations (joins, aggregations) that will definitely be reused, not after cheap reads.
• Materialize intentionally. Call count() immediately after .cache() to trigger materialization in one place, making pipeline behavior explicit and predictable.
• Unpersist diligently. Always call unpersist() when done with cached data, especially in loops or long-running applications. Memory not freed is memory denied to the next operation.
• Profile before caching. Use Spark's UI to confirm that an operation is a bottleneck and that the cached data is reused. Caching the wrong thing defeats its purpose.
• Consider serialization. Use Kryo or columnar formats to reduce memory footprint, especially for large DataFrames destined for cache.