How do I cache a DataFrame effectively?

Summary

• Cache a DataFrame only when several actions reuse it. cache() and persist() are lazy: Spark stores the data the first time it is computed.
• In PySpark, cache() uses MEMORY_AND_DISK_DESER. Pass a StorageLevel to persist() to choose another level, such as MEMORY_ONLY or a _2 level that replicates partitions.
• persist() only assigns a level to a DataFrame that has none, so call unpersist() before you change it. Cached data is shared across all Spark sessions on the cluster, so unpersisting affects every session.
• Check the result with df.storageLevel and the Storage tab of the Spark UI, which lists a DataFrame only after an action has materialized it.
• On Databricks, serverless compute raises an exception on the DataFrame and SQL cache APIs, and Databricks recommends automatic disk caching for data reads.

How do I cache a DataFrame effectively?

Cache a DataFrame that several actions reuse, choose a storage level that fits your memory, confirm in the Spark UI that it materialized, and release it with unpersist() when you are done. The Spark RDD guide calls caching "a key tool for iterative algorithms and fast interactive use."

from pyspark.storagelevel import StorageLevel

df = spark.range(1000).selectExpr("id", "id % 10 AS bucket")
df.persist(StorageLevel.MEMORY_AND_DISK)

df.count()                           # the first action stores the data
df.groupBy("bucket").count().show()  # later actions reuse it
df.unpersist()

What do cache() and persist() do?

DataFrame.cache() persists the DataFrame with the default storage level, MEMORY_AND_DISK_DESER. DataFrame.persist() takes a StorageLevel and stores the contents "after the first time it is computed", so neither call computes anything on its own. persist() "can only be used to assign a new storage level if the DataFrame does not have a storage level set yet." Cached data is shared across all Spark sessions on the cluster.

Which storage level should I choose?

Storage level	Behavior
MEMORY_AND_DISK_DESER	Default for cache() and persist() on a DataFrame
MEMORY_ONLY	Partitions that do not fit in memory are not cached and are recomputed each time they are needed
MEMORY_AND_DISK	Partitions that do not fit in memory are stored on disk and read from there
DISK_ONLY	Partitions are stored only on disk
MEMORY_ONLY_2, MEMORY_AND_DISK_2, DISK_ONLY_2	The base level, with each partition replicated on two cluster nodes
OFF_HEAP	Experimental; requires off-heap memory to be enabled

Two pieces of advice from the Spark RDD guide apply here. Do not spill to disk unless the functions that computed your dataset are expensive or filter a large amount of the data, because recomputing a partition may be as fast as reading it from disk. Use a replicated level if you want fast fault recovery: every level recovers lost data by recomputing it, and replicas let tasks continue without waiting.

When should I cache?

Cache a DataFrame that is expensive to compute and read by more than one action. Skip it when only one action reads the data. Storage competes with execution for memory. Execution may evict cached blocks until storage falls under a threshold, and Spark drops old partitions in least-recently-used order. spark.memory.fraction (default 0.6) sizes the shared region and spark.memory.storageFraction (default 0.5) sets the part of it where cached blocks are immune to eviction by execution. The tuning guide's garbage collection advice puts it this way: "it is better to cache fewer objects than to slow down task execution."

How do I check, refresh and release a cache?

• df.storageLevel returns the DataFrame's current storage level.
• The Spark UI Storage tab, at http://<driver-host>:4040 by default, shows storage level, size and partitions, but only after an action has materialized the DataFrame.
• df.unpersist() removes all blocks for the DataFrame from memory and disk. It does not block by default (blocking=False).
• For tables, use spark.catalog.cacheTable("table_name"), spark.catalog.isCached("table_name") and spark.catalog.uncacheTable("table_name"). REFRESH TABLE invalidates the cached entries, which are repopulated lazily.

SQL CACHE TABLE caches immediately unless you add LAZY, and its default storage level is MEMORY_AND_DISK:

CACHE TABLE testCache OPTIONS ('storageLevel' 'DISK_ONLY') SELECT * FROM testData;

How do checkpoints differ from caching?

checkpoint() and localCheckpoint() truncate the logical plan, which the docs call "especially useful in iterative algorithms where the plan may grow exponentially." checkpoint() saves to files inside the directory set with SparkContext.setCheckpointDir() or spark.checkpoint.dir. localCheckpoint() stores data in the executors using the caching subsystem, so those checkpoints "are not reliable." Both APIs are marked experimental.

What mistakes should I avoid?

Mistake	Fix
Expecting cache() to run immediately	Run an action; the data is stored the first time it is computed
Calling persist() with a new level on a persisted DataFrame	Call unpersist() first
Assuming unpersist() only affects your session	Cached data is shared across sessions, so it affects all of them
Looking for the DataFrame in the Storage tab before any action	The tab lists it only after it is materialized

How does Databricks handle caching?

The Databricks disk cache copies remote Parquet data files, including Delta Lake tables, to local storage on workers. Data is cached automatically whenever a file has to be fetched from a remote location, and Databricks recommends using automatic disk caching. The Apache Spark cache is triggered manually and requires code changes. When a worker is decommissioned, the Spark cache stored on it is lost and Spark rereads the missing partitions from source.
Serverless compute does not support the DataFrame and SQL cache APIs. These raise an exception: df.cache(), df.persist(), df.unpersist(), df.checkpoint(), spark.catalog.cacheTable(), spark.catalog.uncacheTable(), spark.catalog.clearCache(), CACHE TABLE, UNCACHE TABLE, REFRESH TABLE and CLEAR CACHE.

FAQs

What is the difference between cache() and persist()?

cache() always uses the default storage level, which is MEMORY_AND_DISK_DESER for a DataFrame. persist() accepts a StorageLevel. For an RDD, cache() is shorthand for MEMORY_ONLY.

Can I cache a SQL table?

Yes. Use spark.catalog.cacheTable("table_name") or CACHE TABLE. Spark SQL caches tables in an in-memory columnar format and scans only the required columns.