TopolearnWithout caching, every action recomputes from the source; with careless caching, executors run out of memory.
Because DataFrames are lazy, each action recomputes the full plan from the source. If you use the same expensive intermediate result several times, for example in several aggregations or an iterative loop, cache it:
df.cache()marks it for caching with the default storage levelMEMORY_AND_DISK(deserialized for DataFrames). It is lazy: the data is stored the first time an action computes it.df.persist(StorageLevel.DISK_ONLY)or other levels such asMEMORY_ONLYchoose explicitly.df.unpersist()frees it. Do this when you're done; cached blocks compete with execution memory.spark.catalog.cacheTable("name")and SQLCACHE TABLEwork on views and tables.
The Storage tab of the Spark UI shows what is cached and how much memory it uses.
When not to cache: data used only once, data read cheaply from Parquet with good pruning, or huge data that would just spill to disk.
Checkpointing (df.checkpoint() after spark.sparkContext.setCheckpointDir(...)) writes the data to reliable storage and truncates the lineage, the chain of transformations Spark would replay on failure. Use it in long iterative jobs where plans grow very large. df.localCheckpoint() is faster but stored on executors, so it isn't fault-tolerant.
Example
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.master("local[2]").getOrCreate()
from pyspark import StorageLevel
base = spark.range(2_000_000).withColumn("v", F.rand(seed=1) * 100)
expensive = base.filter("v > 50").withColumn("bucket", (F.col("v") / 10).cast("int"))
expensive.cache() # lazy: nothing stored yet
print(expensive.is_cached) # True
print(expensive.count() > 0) # True (computes and fills the cache)
print(expensive.storageLevel) # Disk Memory Deserialized 1x Replicated
stats = expensive.groupBy("bucket").count().orderBy("bucket").collect()
print(len(stats)) # 5 (served from the cache)
expensive.unpersist()
spark.sparkContext.setCheckpointDir("ckpt")
short = expensive.checkpoint() # writes data, cuts the lineage
print("Filter" in short._jdf.queryExecution().logical().toString()) # FalseRemember
cache()= lazypersist(MEMORY_AND_DISK); filled by the first action- Cache only data reused several times, and
unpersist()afterwards - Checkpoints truncate lineage in long iterative jobs
Watch out for
- Caching every intermediate DataFrame 'for speed' and starving execution memory.