Keep It Around

The Daily Grind's cleaned, joined, deduplicated order table is built once and then used by five reports. Without help, Spark builds it five times. Every action re-runs the whole lineage from the files, because a DataFrame is a description, not a box of rows (Nothing Runs Until You Ask). cache is how you tell Spark to keep the rows this once.

What it does

Spark's docs, on persisting: "each node stores any partitions of it that it computes in memory and reuses them in other actions on that dataset." It is lazy like everything else: "The first time it is computed in an action, it will be kept in memory on the nodes." So df.cache() on its own does nothing; the next action builds the DataFrame and fills the cache; the action after that is the fast one.

Watch it happen

Four customers and six orders, one of them entered twice and one from a customer nobody has:

clean = (orders.join(customers, "customer_id")
               .dropDuplicates(["order_id"])
               .cache())
clean.count()                              # the first action builds it and fills the cache
result = clean.groupBy("city").count()     # the second action reads the cache

The join and the dedupe ran once, for count(). The groupBy read the kept rows. When the reports are done:

clean.unpersist()

What it costs

Memory, out of the storage share of each executor's unified region (Where the Memory Goes, next). For a DataFrame, PySpark's cache() "Persists the DataFrame with the default storage level (MEMORY_AND_DISK_DESER)": in memory, deserialised, and partitions that do not fit go to local disk rather than being dropped. (The RDD API's cache() defaults to MEMORY_ONLY, and the docs' table of levels is about RDDs, which trips people up.) Spark "automatically monitors cache usage on each node and drops out old data partitions in a least-recently-used (LRU) fashion", so a cache you forgot about can evict one you need. unpersist() when you are done. And a cache used by one action is a pure cost: you paid to store rows you read once.

What this site shows

cache() is accepted here: afterwards clean.is_cached is True and clean.storageLevel prints a level, in Sail's own words. The plan is the same before and after, because one process holding eleven rows has nothing to keep, and the sandbox does not expose pyspark.StorageLevel, so other levels cannot be tried here. On Spark the physical plan shows InMemoryTableScan once a cached DataFrame is read, and the Spark UI's Storage tab lists what is cached and how big it is (Where to Look When It's Slow).

The symptom, and the fix

An iterative job, a loop over months or a model trained in rounds, gets slower each round, and the Stages tab shows the same early stages running again every time. Cache the DataFrame the loop reads, before the loop; unpersist after it. The opposite symptom: executors spilling and evicting while nothing is reused. That is a cache that should not be there.

Self-check

df.cache(), then df.show(), then df.count(): which call is the slow one?

show. It is the first action, so it builds the lineage and fills the cache. count reads it. cache itself took no time at all.

Nothing to grade here. Next, Where the Memory Goes: the region that cache shares with shuffles and sorts, and whose memory an out-of-memory error is talking about.