Caching & Persistence

By the end of this lesson you will be able to decide when caching a DataFrame helps, use cache() and persist() with a storage level, see a cache in the plan and the Spark UI, free it with unpersist(), and explain how checkpointing differs.

From Lazy Evaluation you know that every action reruns the whole chain of transformations from the source. If you use the same cleaned DataFrame for five reports, Spark reads and cleans the raw data five times. Caching keeps the result in executor memory (and disk) after the first time, so later actions reuse it.

Step 1: see the recomputation problem

This Python simulation counts how often the expensive "read and clean" step runs, with and without a cache.

calls = {"clean": 0}

def read_and_clean():
    calls["clean"] += 1                # pretend this reads 500 GB and cleans it
    return [x for x in [1299, 499, -5, 2450, 89] if x > 0]

def report(name, data_source):
    data = data_source()
    print(f"{name:12} -> {sum(data)}")

# no cache: every report rebuilds the data
for name in ["daily total", "by city", "by category"]:
    report(name, read_and_clean)
print("clean ran", calls["clean"], "times without cache\n")

# cache: build once, reuse
calls["clean"] = 0
cached = read_and_clean()
for name in ["daily total", "by city", "by category"]:
    report(name, lambda: cached)
print("clean ran", calls["clean"], "time with cache")

daily total  -> 4337
by city      -> 4337
by category  -> 4337
clean ran 3 times without cache

daily total  -> 4337
by city      -> 4337
by category  -> 4337
clean ran 1 time with cache

Think of a teacher preparing three different worksheets from the same chapter. Without notes, she rereads the whole chapter for each sheet. With notes made once, each sheet is quick.

Step 2: cache a DataFrame in Spark

from pyspark.sql import functions as F

clean = (
    spark.read.csv("orders.csv", header=True, inferSchema=True)
    .filter(F.col("status") == "paid")
    .withColumn("amount_with_gst", F.round(F.col("amount") * 1.18, 2))
)
clean.cache()                 # marks it; nothing is stored yet
print(clean.storageLevel)    # after cache()

clean.count()                 # first action fills the cache
clean.groupBy("city").sum("amount_with_gst").show()      # reads from the cache
clean.groupBy("category").count().show()                 # reads from the cache

Disk Memory Deserialized 1x Replicated

Two things to notice:

• cache() is lazy, like a transformation. It only marks the DataFrame. The first action (count() here) computes it and stores the partitions on the executors.
• The storage level printed is memory and disk, deserialized, one copy. For DataFrames, cache() uses the level MEMORY_AND_DISK_DESER: partitions stay in memory as ready-to-use objects, and those that do not fit spill to local disk instead of being recomputed.

After caching, the plan reads from memory instead of the file. This is the real plan of clean.groupBy("city").count() from our Spark 4.3.0 run, shortened:

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[city#40], functions=[count(1)])
   +- Exchange hashpartitioning(city#40, 200), ENSURE_REQUIREMENTS, [plan_id=318]
      +- HashAggregate(keys=[city#40], functions=[partial_count(1)])
         +- InMemoryTableScan [city#40]
               +- InMemoryRelation [order_id#38, customer#39, city#40, ...], StorageLevel(disk, memory, deserialized, 1 replicas)
                     +- *(1) Project [...]
                        +- *(1) Filter (isnotnull(status#43) AND (status#43 = paid))
                           +- FileScan csv [...]

InMemoryTableScan reads the cached partitions; the file scan below it runs only if a cached partition is missing. The Storage tab of the Spark UI lists the cached DataFrame, the fraction in memory and the size (Reading the Spark UI).

Step 3: choose a storage level with persist()

persist(level) is cache() with a choice of where to keep the data:

from pyspark import StorageLevel

clean.persist(StorageLevel.MEMORY_AND_DISK)

Serialized means stored as compact bytes, which saves memory but costs CPU to read; deserialized means stored as ready-to-use objects. When in doubt, use plain cache().

Step 4: free memory with unpersist()

Cached data stays until the application ends or memory pressure evicts it (least recently used first). When you are done with it, release it:

clean.unpersist()
spark.catalog.clearCache()     # removes every cached table and DataFrame

Forgetting this in long notebooks is common. The executors fill up with old caches, and new work spills to disk or fails.

When caching helps, and when it hurts

Cache when:

• the same DataFrame feeds several actions (many reports, many model training passes);
• it comes from expensive work (big joins, heavy parsing) and is much smaller than the source;
• you run iterative algorithms that loop over the same data, such as machine learning.

Do not cache when:

• the DataFrame is used only once: caching costs time and memory for nothing;
• it is huge compared to cluster memory: it spills to disk and pushes out more useful data;
• the source is already fast and columnar (Parquet on local SSD) and you read few columns.

A quick estimate before caching:

def worth_caching(uses, compute_minutes, cache_minutes, read_cache_minutes):
    without = uses * compute_minutes
    with_cache = compute_minutes + cache_minutes + (uses - 1) * read_cache_minutes
    return without, with_cache

for uses in [1, 2, 5]:
    without, with_cache = worth_caching(uses, compute_minutes=12, cache_minutes=3, read_cache_minutes=1)
    verdict = "cache it" if with_cache < without else "do not cache"
    print(f"{uses} use(s): {without:>3} min without, {with_cache:>3} min with -> {verdict}")

1 use(s):  12 min without,  15 min with -> do not cache
2 use(s):  24 min without,  16 min with -> cache it
5 use(s):  60 min without,  19 min with -> cache it

Checkpointing: cutting the lineage

Caching keeps the lineage: if an executor dies, Spark can still rebuild lost cached partitions from the source. A checkpoint instead writes the data to reliable storage (HDFS, S3) and forgets the lineage. Use it when the lineage becomes very long, for example in a loop that updates a DataFrame 100 times, where the plan itself grows huge and slow to optimise.

spark.sparkContext.setCheckpointDir("hdfs:///tmp/checkpoints")
long_chain = long_chain.checkpoint()      # writes now and returns a DataFrame with a short plan

localCheckpoint() is a faster variant that keeps data on executors' local storage, but it loses fault tolerance if an executor dies. Structured Streaming uses a different kind of checkpoint, a folder that remembers progress; you will see it in Structured Streaming Basics.

Common mistakes

• Calling cache() and expecting it to run. Nothing is stored until an action.
• Caching everything "to be safe". Memory fills up and Spark spills or evicts useful data.
• Caching before filtering. Filter and select first, then cache the smaller result.
• Never calling unpersist() in long-running notebooks or jobs.

Key points

• Without caching, every action recomputes the full lineage from the source.
• cache() is lazy and uses MEMORY_AND_DISK_DESER for DataFrames; the first action fills it.
• persist(StorageLevel...) chooses memory, disk, serialized or replicated storage.
• Cache data that is reused several times and expensive to compute; release it with unpersist().
• checkpoint() saves data to reliable storage and cuts a very long lineage.

def worth_caching(uses, compute_minutes, cache_minutes, read_cache_minutes):
    without = uses * compute_minutes
    with_cache = compute_minutes + cache_minutes + (uses - 1) * read_cache_minutes
    return without, with_cache

for uses in range(1, 7):
    without, with_cache = worth_caching(uses, 4, 5, 2)
    print(uses, without, with_cache, "cache it" if with_cache < without else "do not cache")

1 4 9 do not cache
2 8 11 do not cache
3 12 13 do not cache
4 16 15 cache it
5 20 17 cache it
6 24 19 cache it