Memory Management and Caching

## What persist actually does `persist(level)` marks a DataFrame or RDD so that the *first* time each of its partitions is computed, the result is kept for reuse. It is **lazy**: the call itself moves no data. Nothing is in memory until an action (`count()`, `write`, `collect()`) forces those partitions to be computed. `cache()` is just `persist()` with the API's default level. Caching happens **per partition**, and per partition it is all-or-nothing: a partition is either fully stored or not stored at all. That is why the Spark UI's *Storage* tab reports a cached fraction — a 200-partition DataFrame can be 63% cached because storage memory filled up part-way through. ## MEMORY_ONLY Partitions are held as deserialized JVM objects (for RDDs) or as Spark's in-memory columnar batches (for DataFrames) in the storage side of the unified memory pool. If a partition does not fit, it is **dropped** — not written anywhere. The next action that needs it re-executes the whole lineage that produced it: re-reading the source files, re-running the filters, and in the worst case re-running an upstream shuffle. This is the level to choose when the upstream work is cheap (a plain file scan) and memory is plentiful; it is the level that quietly turns a "cached" pipeline into one that recomputes on every action. ## MEMORY_AND_DISK Same behaviour while memory lasts; when a partition will not fit, or is later evicted because execution memory reclaimed borrowed space, the block is **written to the executor's local disk directories** (`spark.local.dir` / the cluster manager's scratch dirs) and read back on demand. You trade a disk read for a recomputation. When the upstream chain contains a shuffle, a join, or an expensive UDF, that trade is almost always worth it — which is exactly why the Dataset API defaults to it. Note what disk here is *not*: it is the executor's local scratch space, not durable storage. If the executor dies, the blocks die with it and the partition is recomputed from lineage on a new executor. ## The SER variants `MEMORY_ONLY_SER` and `MEMORY_AND_DISK_SER` keep blocks **serialized** as byte arrays. That typically cuts the footprint substantially and gives the garbage collector one big object instead of millions of small ones, at the cost of CPU on every read. Two caveats worth knowing: - In **PySpark**, storage levels are serialized regardless — Python objects are pickled — so the SER distinction does not apply the way it does in Scala. - For **DataFrames/Datasets**, cached data already uses Spark SQL's compact columnar in-memory format (compressed by default via `spark.sql.inMemoryColumnarStorage.compressed`), so reaching for the `_SER` level buys much less than it does for a raw RDD of case-class objects. ## The rest of the family `DISK_ONLY` skips memory entirely — useful when a result is large, reused a few times, and expensive to compute, but you do not want it competing with execution memory. `OFF_HEAP` stores blocks serialized in the off-heap pool, and requires `spark.memory.offHeap.enabled`. The `_2` suffixed variants (`MEMORY_AND_DISK_2`, etc.) replicate every block on two executors, so losing one executor does not force recomputation. Replication doubles the memory cost and is rarely justified for batch work. ## When caching is the wrong answer Caching earns its cost only when a dataset is read **more than once**. A single linear pipeline that reads, transforms and writes once gains nothing from `cache()` and loses storage memory that execution could have used. Caching also blocks some optimizations: once a plan is cached, downstream filters cannot be pushed through the cache barrier into the source scan, so a `cache()` placed too early can make a job read far more data than the uncached plan would have. ## Releasing the memory `unpersist()` releases the blocks; by default it is non-blocking, and it takes an optional `blocking` flag. Spark's `ContextCleaner` will eventually free cached blocks whose RDD or DataFrame has been garbage-collected on the driver, so short-lived batch applications often get away without calling it. Long-running applications — notebooks, streaming drivers, Thrift servers — must unpersist explicitly, or storage memory slowly fills with datasets nobody references and execution memory starves. ## Reading the evidence The Storage tab shows, per cached dataset, the storage level, the number of cached partitions, the fraction cached, and the split between size in memory and size on disk. A large "size on disk" number next to a `MEMORY_AND_DISK` level tells you memory is the constraint; a cached fraction well under 100% on `MEMORY_ONLY` tells you your job is silently recomputing.