Definition
- unpersist() – removes a dataset's cached partitions from all executors, freeing memory and disk immediately (unlike cache, which is lazy).
- Why important – cached data lingers until evicted or explicitly removed; in long-running applications, successive caches accumulate and exhaust storage, causing eviction and slowdowns.
- Monitoring – the Spark UI Storage tab lists each cached dataset with partition count, fraction cached, memory used, and disk used.
- Signals – fraction cached below 100% means data did not fit; rising eviction means caches should be unpersisted earlier.
Picture the clerk at the National Bank of India who keeps finished summaries on his desk. Unpersist is clearing a summary off the desk once the reports that needed it are done. In a short job this barely matters, but in a long shift where he keeps producing summaries, if he never clears the old ones the desk overflows and important papers fall off. The Storage tab is his dashboard showing which summaries are on the desk, how much space each takes, and whether they are fully laid out.
Understanding the Concepts
How does unpersist() work?
unpersist() tells the block manager to drop all cached blocks for a dataset across every executor. Unlike cache(), which is lazy and only marks intent, unpersist() acts promptly. After calling it, the memory and disk space are immediately available for other caches or execution.
from pyspark.sql import SparkSession, functions as F
spark = SparkSession.builder.appName("NBI-Unpersist").getOrCreate()
txns = spark.read.parquet("s3://nbi-data-lake/transactions/")
cleaned = txns.filter(F.col("status") == "SETTLED")
# Cache and materialize
cleaned.cache()
cleaned.count()
print(f"Is cached: {cleaned.is_cached}") # True
# Release the cache
cleaned.unpersist()
print(f"Is cached: {cleaned.is_cached}") # False
# Memory is now free for other caches or execution
Is cached: True
Is cached: False
Why is unpersist important in long-running applications?
A short batch job ends and releases everything, so lingering caches are harmless. But a long pipeline, a streaming application, or an interactive session runs for hours. If each stage caches new datasets without releasing old ones, cached data accumulates until storage is full. At that point, executors start evicting useful partitions.
# WRONG: long pipeline without unpersist
# Phase 1
phase1_data = compute_phase1(txns)
phase1_data.cache()
phase1_data.count()
# ... generate phase 1 reports ...
# Phase 2 (phase1 cache still holding memory!)
phase2_data = compute_phase2(txns)
phase2_data.cache()
phase2_data.count()
# Storage is now overloaded. Phase 2 evicts phase 1 partitions,
# and if phase 1 is still needed, those are recomputed.
# RIGHT: unpersist between phases
phase1_data = compute_phase1(txns)
phase1_data.cache()
phase1_data.count()
# ... generate phase 1 reports ...
phase1_data.unpersist() # Free memory BEFORE phase 2
phase2_data = compute_phase2(txns)
phase2_data.cache()
phase2_data.count()
# Full memory available for phase 2
How do you monitor caching in the Spark UI?
The Storage tab lists every cached RDD or DataFrame with concrete numbers.
# After caching and running actions, navigate to:
# http://driver-host:4040/storage/
# You will see:
# +-------------------+-------+--------+--------+--------+
# | RDD Name | Level | Cached | Memory | Disk |
# | | | Parts | Size | Size |
# +-------------------+-------+--------+--------+--------+
# | cleaned | M+D | 200/200| 2.4 GB | 0 B |
# | phase2_data | M+D | 150/200| 1.8 GB | 600 MB |
# +-------------------+-------+--------+--------+--------+
# Key metrics to check:
# - Cached Partitions: should equal total partitions
# - Fraction Cached: should be 100%
# - Size in Memory vs Size on Disk
cleaned.cache()
cleaned.count()
print(f"Storage level: {cleaned.storageLevel}")
Storage level: Disk Memory Deserialized 1x Replicated
What does partial caching mean?
If the Storage tab shows a dataset is only partially cached (fraction below 100%), some partitions did not fit in memory. With MEMORY_AND_DISK, those spilled to disk. With MEMORY_ONLY, they were dropped and must be recomputed on access.
# Storage tab shows: cleaned - 120/200 partitions cached (60%)
# This means 80 partitions were evicted or did not fit.
# Fix options:
# 1. Free other caches to make room
other_cache.unpersist()
# 2. Switch to serialized level to fit more data
from pyspark import StorageLevel
cleaned.unpersist()
cleaned.persist(StorageLevel.MEMORY_AND_DISK_SER)
cleaned.count()
# 3. Increase executor memory
# spark-submit --executor-memory 16g ...
How do you use the Executors tab for cache monitoring?
The Executors tab shows each executor's memory usage, including how much is used for storage (caching) versus execution (shuffles, sorts).
# Navigate to: http://driver-host:4040/executors/
# For each executor you see:
# - Storage Memory: used / total (e.g., 2.1 GB / 4.0 GB)
# - If storage is near capacity, caches are at risk of eviction
# Tip: if one executor is much more loaded than others,
# the cached data is skewed and some partitions are larger.
How do you use the Stages tab to verify cache hits?
The Stages tab shows task-level details that reveal whether cached data is actually being read.
# Navigate to: http://driver-host:4040/stages/
# Click on a stage that reads cached data.
# For each task, check:
# - Input Size: should show "In Memory" or "On Disk" (cached)
# - Duration: cached tasks should be much faster than recomputed ones
# If tasks show "HDFS" or "S3" as input, they are reading from
# source, not cache. The cache was evicted or never materialized.
What are the blocking vs non-blocking variants?
unpersist() has a blocking parameter. By default it is non-blocking (asynchronous). Passing blocking=True makes it wait until all blocks are removed.
# Non-blocking (default): returns immediately, cleanup in background
cleaned.unpersist() # or unpersist(blocking=False)
# Blocking: waits until all cached blocks are freed
cleaned.unpersist(blocking=True)
# Useful when you need to guarantee memory is free before next cache
# Example: ensure phase 1 is fully freed before phase 2
phase1_data.unpersist(blocking=True)
phase2_data.cache()
phase2_data.count()
What should a senior engineer know?
A senior engineer reads the Storage, Executors, and Stages tabs together to tune caching. Rising eviction counts or growing GC time over a long job point to caches that should have been unpersisted. A cached dataset that is never read again is wasted memory.
# Senior monitoring checklist:
# 1. Storage tab: all caches at 100% fraction?
# 2. Executors tab: storage memory not at capacity?
# 3. Stages tab: cached reads are fast?
# 4. GC metrics: GC time not increasing over the job?
# If any check fails:
# - Unpersist unused caches
# - Switch to serialized storage level
# - Increase executor memory
# - Reduce what you cache
Visual Overview
Cache lifecycle in a long pipeline:
Phase 1: cache() → count() → reports → unpersist()
▲ mark ▲ store ▲ read ▲ free
Phase 2: cache() → count() → reports → unpersist()
▲ mark ▲ store ▲ read ▲ free
Memory over time (with unpersist):
┌────────────────────────────────────────┐
│ Storage Memory │
│ ████████░░░░░░░░░░████████░░░░░░░░░░ │
│ Phase 1 free Phase 2 free │
└────────────────────────────────────────┘
Memory over time (without unpersist):
┌────────────────────────────────────────┐
│ Storage Memory │
│ ████████████████████████████████ FULL! │
│ Phase 1 + Phase 2 → eviction! │
└────────────────────────────────────────┘
Key Takeaway
The unpersist() method removes a dataset's cached partitions from every executor and frees memory immediately. Unlike cache(), which is lazy, unpersist() acts promptly so the storage is available right away. This distinction matters enormously in long-running applications.
In a short batch job, lingering caches are harmless because the job ends and releases everything. But in a long pipeline, a streaming application, or an interactive session, caches from earlier phases accumulate. If you never release them, they fill the storage region, and executors start evicting useful partitions from later phases. Calling unpersist() as soon as a dataset's reuse phase is finished keeps the storage footprint bounded.
Monitoring is how you know caching is working. The Spark UI Storage tab shows each cached dataset's partition count, fraction cached, and memory and disk usage. A dataset at less than 100% fraction means it did not fit. The Executors tab shows storage memory per executor. The Stages tab reveals whether tasks are reading from cache or recomputing from source.
At the National Bank of India our long multi-phase nightly pipeline unpersisted each phase's caches before the next phase began, and we checked the Storage tab to confirm the shared cleaned dataset was fully cached and hit by every report.
Remember one sentence: unpersist() promptly frees cache memory, which is essential in long-running applications to prevent storage exhaustion and eviction of useful cached data.
---