Spark Cache Strategy, in depth: counting reuse, a cost model, where to cut the cache point, lifecycle and staleness

The architecture

What a cache actually buys

Spark evaluates lazily. A DataFrame is a plan, not data, and each action (count, write, collect, toPandas) compiles and runs the plan from its sources. If four actions share a prefix, the prefix runs four times: four scans of the same files, four evaluations of the same filters, four shuffles if the prefix contains a join.

A cache changes this by registering the plan with the session's cache manager. The first action that touches it computes the result and stores it partition by partition; later queries whose plans contain an equivalent subtree read the stored partitions instead. Equivalence is checked on the analysed plan, not on the Python variable, so a later query built on top of the cached DataFrame hits the cache, while a query that re-derives the same data with a different filter order or an extra column may not. The practical rule: build downstream work from the cached object itself.

So the question a cache answers is narrow. Is there a prefix of the plan that is expensive, reused by at least two actions, small enough after filtering to fit in the memory you can spare, and stable for as long as you hold it? If any of the four is false, caching is the wrong tool.

Counting reuse before you cache

Do not guess reuse; read it. Two sources tell you. The first is the code: list every action in the job and trace each back to its sources. Any DataFrame that two or more actions depend on is a candidate. The second is the SQL tab of the Spark UI: the same file scan with the same pushed filters appearing in several queries is reuse the code did not make obvious, often because a helper function rebuilds the DataFrame each time it is called.

Then confirm the hit. Explain plans show an InMemoryTableScan where a cache serves the subtree; a FileScan in the same place means the lookup missed and you are paying for the cache and the recompute.

# Find reuse before deciding: which plan prefixes do several actions replay?
# Run each candidate action with explain() and look for the same scan and filter.
daily.explain("formatted")
anomalies.explain("formatted")

# After caching, the formatted plan should show InMemoryRelation / InMemoryTableScan
# where the scan used to be. If it still shows FileScan, the lookup missed.
base = spark.read.parquet("s3://lake/events/").where("event_date >= '2026-09-01'")
base = base.select("user_id", "event_type", "amount", "country", "ts")
base.cache()
base.groupBy("country").sum("amount").explain()   # expect InMemoryTableScan

A cost model you can compute

The benefit is simple arithmetic once you measure three times. Recompute time is how long the shared prefix takes from source; read it from the stage durations of one uncached run. Materialise time is the first computation plus the work of encoding the result into cached batches, usually somewhat more than one recompute. Cached read time is how long a consumer takes to scan the cached data, which is small when it sits in memory and larger when it has spilled to local disk.

def cache_benefit(recompute_s, consumers, materialise_s, cached_read_s):
    """Seconds saved by caching a prefix that `consumers` actions share.

    recompute_s:   time to compute the prefix from source once (from the UI)
    materialise_s: first computation plus the cost of building the cache
    cached_read_s: time for one consumer to read the cached result
    """
    without_cache = consumers * recompute_s
    with_cache = materialise_s + consumers * cached_read_s
    return without_cache - with_cache

# A positive result is necessary, not sufficient: also check the memory it displaces.

Two costs do not appear in the formula and decide many real cases. The first is displacement: storage memory and execution memory share one unified region, and execution can evict cached blocks when a join or sort needs room (the mechanics page walks through the eviction rules). A large cache can turn a fast in-memory sort into a spilling one, and that slowdown lands on stages that never touch the cache. The second is the single consumer case: with one action, caching always loses, because you pay materialisation on top of the one computation you needed anyway.

Where to cut the cache point

Where you place the cache matters as much as whether you cache. Three rules cover most pipelines.

1. Cut after the narrowing. Apply filters and column projection before cache(). Caching the raw table and filtering afterwards stores every column of every row, so the cache may be twenty times larger than the data the consumers read. Columnar cached batches do not save you here: unused columns are still materialised.
2. Cut before the fan-out. The cache point should be the last node every consumer shares. Cut earlier and each consumer repeats work after the cache; cut later and some consumers cannot use it.
3. Cut after expensive shuffles when they are shared. A join of two large tables that three reports reuse is the classic case. A cache after the join stores the joined rows once; without it the shuffle runs three times.

A useful anti-rule: do not cache inside a loop that redefines the DataFrame each iteration without unpersisting the previous one. Each iteration registers a new cache entry and the old ones are only released when evicted, which is how notebooks end up with dozens of orphaned entries.

Lifecycle: materialise, use, release

Treat a cache like a file handle: something opens it, something closes it, and the two are in the same function. cache() is lazy, so the first consumer pays the materialisation and its timing is distorted; an explicit count() moves that cost to a known place and makes the cache complete before consumers run. If a cached partition is evicted or its executor is lost, Spark recomputes it from lineage. Without external writes that costs only speed; after one, a partially evicted cache can return a mix of old and new rows, which is another reason to refresh, as the staleness section explains.

from contextlib import contextmanager
from pyspark import StorageLevel

@contextmanager
def cached(df, level=StorageLevel.MEMORY_AND_DISK):
    """Cache for the duration of a block, materialise once, always release."""
    df = df.persist(level)
    df.count()                      # materialise deliberately, not inside consumer 1
    try:
        yield df
    finally:
        df.unpersist(blocking=False)

with cached(base) as b:
    b.groupBy("country").sum("amount").write.mode("overwrite").parquet(out_rollup)
    b.where("amount > 10000").write.mode("overwrite").parquet(out_anomaly)
    b.select("user_id", "ts").write.mode("overwrite").parquet(out_export)

Release with unpersist() when the last consumer is done. blocking=False returns immediately and lets executors free blocks in the background; use blocking=True when the next step needs the memory at once. In SQL the equivalents are CACHE TABLE and UNCACHE TABLE IF EXISTS. Note that CACHE TABLE is eager by default and scans the table immediately; CACHE LAZY TABLE defers the work to the first query, which is what you want in a script that may not reach the consumers.

Staleness and shared sessions

A cache stores the data as it was when the cache was built. That is the point, and it is also the most common source of wrong answers. When the same Spark application writes to a table through the catalog, Spark invalidates cache entries that depend on that table. When anything else writes to the underlying files (another job, another cluster, an ingestion service), the cache does not know, and queries keep returning the old rows until you refresh.

• REFRESH TABLE db.events or spark.catalog.refreshTable("db.events") invalidates cached data and metadata for a catalog table.
• spark.catalog.refreshByPath("s3://lake/events/") does the same for path-based reads.
• For a long-running service that caches a dimension table, schedule the refresh explicitly; do not rely on eviction to happen to coincide with upstream writes.

The cache manager also lives in state shared by every session of one Spark application. A notebook kernel, a Thrift server or a long-running app with several SparkSession objects shares cache entries across them, and spark.catalog.clearCache() called by one user removes entries everyone else was relying on. On shared clusters, prefer targeted UNCACHE TABLE over clearing everything.

When not to cache

Several situations look like they need a cache and do not.

• Reuse inside one query. When the same subplan appears twice in one query, for example a self-join or a union of two aggregations over the same scan, Spark's exchange reuse can compute an identical shuffle or broadcast once and read it twice. Check the plan for ReusedExchange before adding a cache.
• A single action. One consumer means caching is pure overhead, as the model shows.
• A cheap source. If recomputation is a scan of a few gigabytes of well-partitioned Parquet with partition pruning, the cached read may barely beat it while costing memory.
• Reuse across jobs. A cache dies with the application. If tomorrow's job needs the result, write a table; the trade-off is set out in Spark Persistence.
• Vendor file caches. Some platforms cache remote file data on local SSDs below Spark (Databricks calls its version the disk cache). That layer is automatic, survives across queries in a cluster and does not consume executor memory; it is vendor-specific and does not replace a Spark cache of a computed result.

The one streaming pattern where caching is standard is foreachBatch with several sinks: without a persist, each write recomputes the micro-batch from the source.

def write_two_sinks(batch_df, batch_id):
    batch_df.persist()
    try:
        batch_df.write.format("delta").mode("append").save("/lake/bronze/events")
        (batch_df.groupBy("country").count()
                 .write.format("delta").mode("append").save("/lake/silver/country_counts"))
    finally:
        batch_df.unpersist()

stream.writeStream.foreachBatch(write_two_sinks).start()

Worked example: one prefix, four consumers

A nightly job reads a month of events (600 GB of Parquet, 80 columns), keeps rows from four countries and twelve columns, and feeds four actions: a rollup, an anomaly scan, an export and an audit count. The following timings are illustrative inputs of the kind you read from the stage view, not a benchmark.

Without a cache the prefix costs 4 x 240 = 960 s. With a memory-resident cache it costs 270 + 4 x 15 = 330 s, saving 630 s. If memory is short and part of the cache spills, it costs 270 + 4 x 45 = 450 s, still a saving of 510 s. With a single consumer the same cache costs 270 + 15 = 285 s against 240 s, a loss of 45 s. Had the job cached the raw scan instead of the filtered prefix, the cache would have held all 80 columns of every row and would likely have displaced the memory the rollup's aggregation needed.

The decision is therefore: cache after the filter and projection, materialise with count(), run the four consumers, unpersist. Then verify the plan of each consumer shows InMemoryTableScan and that the Storage tab reports the fraction cached at 100 percent rather than a partial figure.

Failure modes

• Silent misses. The consumer re-derives the data rather than building on the cached object; the plan shows a file scan and you pay twice. Check the plan, not the code.
• Orphaned entries. A loop or a retried notebook cell caches new DataFrames without unpersisting old ones. The Storage tab lists entries nobody references.
• Stale answers. An external writer updates the source; the cache keeps serving yesterday's rows. Refresh after upstream writes or do not hold the cache across them.
• Partial caching. Not every partition fits; the uncached ones are recomputed on each access, so the job pays most of the recompute cost plus the cache. The Storage tab's fraction cached shows it.
• Execution starvation. A large cache leaves joins and sorts too little room, and they spill. Watch spill metrics on stages that do not read the cache.
• Shared-session surprises. clearCache() in one notebook empties another's cache on the same application.

Trade-offs

What to do next

1. List every action in your heaviest job and mark DataFrames used by two or more of them.
2. For each candidate, read recompute time from one uncached run and compute the benefit with the model above; drop candidates with one consumer.
3. Move each cache point after filters and projection and before the fan-out.
4. Wrap caches in a helper that materialises with count() and always unpersists.
5. Confirm InMemoryTableScan in each consumer's plan and 100 percent cached in the Storage tab.
6. Add REFRESH TABLE or refreshByPath wherever an external writer can change a cached source, and replace clearCache() on shared clusters with targeted uncache calls.