How do I cache a DataFrame, and when should I?

Summary

• Call df.cache() or df.persist(level). Both are lazy: Spark stores the data the first time an action computes it, and later actions reuse it.
• In PySpark, cache() uses the default storage level MEMORY_AND_DISK_DESER. persist() takes a StorageLevel such as MEMORY_ONLY or DISK_ONLY.
• Cache a DataFrame that several actions reuse and that is costly to recompute. Skip single-use or cheap results.
• Free the cache with unpersist(). Cached data is shared across all Spark sessions on the cluster.
• On Databricks, serverless compute doesn't support the DataFrame and SQL cache APIs, and Databricks recommends its automatic disk cache.

How do I cache a DataFrame, and when should I?

Call cache() or persist() on a DataFrame you plan to reuse. Caching is lazy: Spark keeps the data the first time an action computes it, then reuses those partitions in later actions on that DataFrame and DataFrames derived from it. Cache when several actions read the same result and recomputing it is costly, and call unpersist() when you're done.

from pyspark.storagelevel import StorageLevel

orders = spark.read.parquet("/data/orders").filter("amount > 100")
orders.cache()                           # marks it; nothing is stored yet
orders.count()                           # first action materializes the cache
orders.groupBy("region").count().show()  # reads the cached partitions
print(orders.storageLevel)

orders.unpersist()                       # removes its blocks from memory and disk
orders.persist(StorageLevel.DISK_ONLY)   # choose a level explicitly

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

In PySpark, cache() persists the DataFrame with the default storage level, MEMORY_AND_DISK_DESER. That default changed in Spark 3.0 to match Scala. persist() accepts an optional StorageLevel and falls back to the same default. It can only assign a level when the DataFrame has none yet, so call unpersist() before you switch levels.

Which storage level should I choose?

Storage level	What it does	When to use it
MEMORY_AND_DISK_DESER	Default for DataFrame.cache() and persist() in PySpark	General reuse
MEMORY_ONLY	Partitions that don't fit aren't cached and are recomputed when needed	The data fits and recomputing a partition is cheap
MEMORY_AND_DISK	Partitions that don't fit go to disk and are read from there	Recomputing is expensive
DISK_ONLY	Stores partitions only on disk	Memory is tight but recomputing is still costly
MEMORY_AND_DISK_2 (and other _2 levels)	Replicates each partition on two cluster nodes	You want fast fault recovery
OFF_HEAP	Experimental; uses off-heap memory	Off-heap memory is enabled

Spark's RDD programming guide advises against spilling to disk unless the functions that computed the data are expensive or filter a large amount of it, since recomputing a partition may be as fast as reading it from disk. Every level is fault tolerant, because Spark rebuilds a lost partition from the transformations that created it. Replicated levels let tasks keep running without waiting for that rebuild.

When does caching pay off?

The RDD guide calls caching "a key tool for iterative algorithms and fast interactive use". It pays off when the same DataFrame feeds several actions or when one result goes to several destinations. The Structured Streaming guide applies this to foreachBatch: persist the micro-batch, write it to each location, then unpersist it, so each write doesn't recompute the output. Skip it for a DataFrame you use once, since only later actions benefit, and for cheap upstream work.

How do I release cached data?

unpersist() marks the DataFrame as non-persistent and removes all of its blocks from memory and disk. It doesn't block by default (blocking=False). Spark also drops old partitions on its own, in least-recently-used (LRU) order. spark.catalog.clearCache() removes all cached tables from the in-memory cache. Cached data is shared across all Spark sessions on the cluster, so unpersisting or clearing it affects every session.

How do I cache a table with SQL?

CACHE TABLE big_orders OPTIONS ('storageLevel' 'MEMORY_ONLY')
  SELECT * FROM orders WHERE amount > 1000;

UNCACHE TABLE IF EXISTS big_orders;
CLEAR CACHE;

CACHE TABLE caches immediately unless you add LAZY, and its default level is MEMORY_AND_DISK. In Python, spark.catalog.cacheTable("orders") caches a named table (the storageLevel argument arrived in Spark 3.5.0), and spark.catalog.isCached("orders") reports whether it is cached. Spark SQL stores cached tables in an in-memory columnar format, scans only the required columns and tunes compression automatically.

How is checkpoint() different from cache()?

A cache keeps the lineage, so Spark can rebuild lost partitions. checkpoint() truncates the logical plan, which helps in iterative algorithms where the plan may grow exponentially. It writes files to the directory set with SparkContext.setCheckpointDir() or the spark.checkpoint.dir configuration. localCheckpoint() keeps the data on executors through the caching subsystem, so it isn't reliable. The docs mark both APIs as experimental.

How can I see what is cached?

The Storage tab of the Spark UI lists persisted RDDs and DataFrames with their storage levels, sizes and partitions. A newly persisted DataFrame shows up there only after an action materializes it. In df.explain(), a cached DataFrame shows an InMemoryTableScan node.

How does Databricks approach DataFrame caching?

Databricks adds a disk cache that copies remote Parquet data files, including Delta Lake tables, into the nodes' local storage on first read, where the cache is enabled. It differs from Apache Spark caching, and Databricks recommends using the automatic disk cache. When a worker is decommissioned, the Spark cache stored on it is lost, so autoscaling makes Spark caching less stable. Serverless compute doesn't support the DataFrame and SQL cache APIs: df.cache(), df.persist(), df.unpersist(), df.checkpoint(), spark.catalog.cacheTable(), uncacheTable(), clearCache(), CACHE TABLE, UNCACHE TABLE and CLEAR CACHE raise an exception there. SQL warehouses also keep a query result cache that is invalidated when the underlying tables are updated.

FAQs

Does cache() store data right away?

No. cache() and persist() only mark the DataFrame, and the data is stored when the first action computes it. CACHE TABLE without LAZY is the eager option.

Why do the default storage levels differ?

Each API documents its own default. RDD.cache() uses MEMORY_ONLY, PySpark's DataFrame.cache() uses MEMORY_AND_DISK_DESER, and Catalog.cacheTable() and CACHE TABLE use MEMORY_AND_DISK.

Can I change the storage level of a cached DataFrame?

Not in place. Call unpersist(), then persist() with the new level.