Spark Caching: cache(), persist(), Storage Levels and Garbage Collection

1.cache() versus persist() and their defaults

Executors are long-lived JVM processes that can keep data in memory or on disk between tasks. Spark lets you use this to keep a dataset around across operations. When you persist a dataset, each node stores the partitions it computes and reuses them in later actions on that dataset or on datasets derived from it. The RDD guide says this often makes later actions more than 10x faster. Caching is a key tool for iterative algorithms and interactive work.

cache() takes no arguments and always uses the default storage level. persist() accepts a StorageLevel, and called with no argument it uses the same default. The default depends on which API you are using:

- DataFrame cache() and persist() default to MEMORY_AND_DISK_DESER. The PySpark default changed to this in 3.0 to match Scala. - RDD cache() is shorthand for StorageLevel.MEMORY_ONLY, which stores deserialized objects in memory.

You can assign a storage level with persist() only if the DataFrame does not already have one.

Neither call does any work immediately. The level takes effect the first time the DataFrame is computed. The data is kept from that first action onward and reused afterwards.

df = spark.range(1)
df.persist()
# DataFrame[id: bigint]

df.explain()
# == Physical Plan ==
# InMemoryTableScan ...

from pyspark.storagelevel import StorageLevel
df.persist(StorageLevel.DISK_ONLY)
# DataFrame[id: bigint]

from pyspark.storagelevel import StorageLevel
df.persist(StorageLevel. ? )

Sources123

2.Storage levels: trading memory against CPU

A storage level decides where persisted partitions live, whether they are serialized and whether they are replicated. The RDD programming guide lists the full set. MEMORY_AND_DISK_DESER, the DataFrame default, is named in the DataFrame reference, but that table does not include it.

In Python the serialized/deserialized distinction does not apply, because Python always serializes stored objects with Pickle. The Python storage levels are MEMORY_ONLY, MEMORY_ONLY_2, MEMORY_AND_DISK, MEMORY_AND_DISK_2, DISK_ONLY, DISK_ONLY_2 and DISK_ONLY_3.

To choose a level, the guide gives a sequence:

1. If the data fits comfortably at MEMORY_ONLY, stay there. It is the most CPU-efficient option. 2. If it does not fit, try a serialized level with a fast serializer (Java and Scala). 3. Spill to disk only when computing the partitions was expensive or filtered out a lot of data. Otherwise recomputing a partition can be as fast as reading it from disk. 4. Use a replicated _2 level when you need fast fault recovery.

Every level is fault-tolerant, because Spark can recompute lost partitions from the transformations that created them. Replication only saves you the wait for that recomputation.

Sources13

3.How long cached data lives, and who can see it

Cached partitions are not permanent. Spark monitors cache usage on each node and drops old partitions in least-recently-used (LRU) order. To free a dataset yourself, call unpersist(). By default it does not block. If you need to wait until the resources are actually freed, pass blocking=true.

The cache is also scoped more widely than one session: cached data is shared across all Spark sessions on the cluster. Cached data is still tied to the executors that hold it. When a worker is decommissioned, for example under autoscaling, its Spark cache is lost, and Spark must reread the missing partitions from the source.

Sources14

4.Executor memory and garbage collection

Memory used for caching and memory used for computation come from the same place: the executor JVM. Databricks describes the executor as a Java process that triggers GC lazily, and notes that the dataset cache uses a lot of executor memory. Large caches therefore reduce the memory left for everything else, which is one reason to unpersist() data you no longer need instead of waiting for LRU eviction.

Shuffles add pressure of their own. Operations such as reduceByKey and aggregateByKey build in-memory structures to organize records. When the data does not fit, Spark spills those tables to disk, which adds disk I/O and increases garbage collection. Shuffles also write intermediate files that are kept until the corresponding RDDs are no longer used and have been garbage collected. If the application keeps references to those RDDs, or GC runs infrequently, cleanup can be delayed for a long time. Long-running jobs can then use a lot of disk space in the directory set by spark.local.dir.

Sources53

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.DataFrame.cache() stores data at MEMORY_ONLY, the same as RDD.cache().Why is that wrong?

   MEMORY_ONLY is the RDD default. For DataFrames, cache() and persist() default to MEMORY_AND_DISK_DESER.

   Covered in cache() versus persist() and their defaults

2. 2.You can call persist() again with a different StorageLevel to change how an already-persisted DataFrame is stored.Why is that wrong?

   persist() can assign a new storage level only when the DataFrame does not have one yet.

   Covered in cache() versus persist() and their defaults

3. 3.unpersist() frees the cached memory before it returns.Why is that wrong?

   unpersist() does not block by default. Pass blocking=true to wait until the resources are freed.

   Covered in How long cached data lives, and who can see it

Practise it for real

See a DataFrame's persistence show up in its physical plan

1. 1.In a PySpark session, run df = spark.range(1)

   Why: range() creates a small DataFrame with a single id column to experiment on

   You should see: A DataFrame with schema [id: bigint]

2. 2.Call df.persist() with no argument

   Why: With no level given, persist() uses the DataFrame default, MEMORY_AND_DISK_DESER

   You should see: DataFrame[id: bigint] is returned

3. 3.Run df.explain()

   Why: The physical plan shows whether the persisted data will be read

   You should see: The plan contains InMemoryTableScan