Module 6: Catalyst Explain And Caching

`persist()` and storage levels

Next →

Description

Lesson 5 used .cache() without explaining a decision Spark makes for you, silently: saving the DataFrame at the MEMORY_AND_DISK_DESER level — in memory, deserialized, with disk as a fallback. .cache() is, in reality, a shortcut: df.cache() is exactly equivalent to df.persist() with no arguments, which in turn uses that level by default. .persist(storageLevel) lets you choose a different level, and this lesson measures, with the same ten million rows, how much memory and how much disk each one takes up — not as a theoretical list of options, but with real, measured bytes.

Connection to the module. Lesson 5 demonstrated caching helps when there's real reuse. This lesson doesn't change that conclusion — it goes deeper into how that result gets saved, because the answer to "should I cache this?" sometimes depends on how much memory you have available, and this lesson gives you the tools to decide with numbers, not with the default option chosen without thinking.

An analogy: three ways to store the same meal

This module's refrigerator doesn't have just one way to store food. You can store it already plated, ready to eat right away — it takes up more space in the refrigerator, but nobody has to do anything before eating — that's storing it deserialized. You can store it vacuum-sealed — it takes up much less space, but you have to open the package and reheat it before serving — that's storing it serialized. And you can store the surplus that doesn't fit in the refrigerator in the basement freezer — nearly unlimited space, but it takes longer to bring up when needed — that's using disk as a fallback. persist(storageLevel) is, precisely, the decision of which of these combinations to use for a specific DataFrame.

Worked example: the same fact, three storage levels, measured

# storage_levels.py
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)
from pyspark import StorageLevel

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").config("spark.driver.memory", "3g").getOrCreate()

scale_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("franchise_id", IntegerType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
orders_at_scale_df = spark.read.csv("kiosko_orders_at_scale.csv", schema=scale_schema, header=True, enforceSchema=False)
fact = orders_at_scale_df.withColumn("revenue", F.col("quantity") * F.col("unit_price"))

def report_rdd_storage():
    infos = spark.sparkContext._jsc.sc().getRDDStorageInfo()
    if len(infos) == 0:
        print("    (no RDD cached)")
    for i in range(len(infos)):
        info = infos[i]
        print(f"    numCachedPartitions={info.numCachedPartitions()} memSize={info.memSize():,} diskSize={info.diskSize():,}")

for level_name in ["MEMORY_ONLY", "MEMORY_AND_DISK", "DISK_ONLY"]:
    level = getattr(StorageLevel, level_name)
    fact.unpersist(blocking=True)   # required -- see "Common mistakes"
    fact.persist(level)
    print(f"\n=== persist({level_name}) -> fact.storageLevel = {fact.storageLevel} ===")
    fact.count()  # materializes
    report_rdd_storage()

fact.unpersist(blocking=True)
spark.stop()

What to expect. Running python3 storage_levels.py, the output is exactly this (executed in this run, over fact_orders_at_scale, ten million rows):

=== persist(MEMORY_ONLY) -> fact.storageLevel = Memory Serialized 1x Replicated ===
    numCachedPartitions=12 memSize=68,215,021 diskSize=0

=== persist(MEMORY_AND_DISK) -> fact.storageLevel = Disk Memory Serialized 1x Replicated ===
    numCachedPartitions=12 memSize=68,215,021 diskSize=0

=== persist(DISK_ONLY) -> fact.storageLevel = Disk Serialized 1x Replicated ===
    numCachedPartitions=12 memSize=0 diskSize=68,215,021

Three results, with the same exact data. MEMORY_ONLY stores all twelve partitions in memory, 68,215,021 bytes (~65 MB), zero on disk. MEMORY_AND_DISK stores the same total, also complete in memory — because, in this run, everything fit without any trouble; disk would only get used as a fallback if available memory fell short. DISK_ONLY stores exactly the same 68,215,021 bytes, but all on disk, zero in memory. Notice something important: the byte size is identical in all three cases — 68,215,021 — because all three levels use the same serialized (compact) encoding; the only thing that changes between them is where that representation lives, not how it's encoded.

Diagram: memory, disk, and each combination's real weight

flowchart TD
    A["fact_orders_at_scale\n10,000,000 rows"] --> B{"storageLevel"}
    B -->|"MEMORY_ONLY"| C["68,215,021 bytes in MEMORY\n(serialized)"]
    B -->|"MEMORY_AND_DISK"| D["68,215,021 bytes in MEMORY\n(disk as fallback if it doesn't fit)"]
    B -->|"DISK_ONLY"| E["68,215,021 bytes on DISK\n(zero in memory)"]
    B -->|"cache() -- default\nMEMORY_AND_DISK_DESER"| F["433,833,928 bytes in MEMORY\n(deserialized, lesson 5)"]

    style C fill:#69c,stroke:#333
    style D fill:#9c6,stroke:#333
    style E fill:#fc6,stroke:#333
    style F fill:#c96,stroke:#333

Going deeper: why 68,215,021 (serialized) and 433,833,928 (deserialized) are such different figures

This contrast isn't an accident of this particular run — it's the underlying reason persist() exists as a separate API from .cache(). Lesson 5 measured 433,833,928 bytes for .cache()'s default level (MEMORY_AND_DISK_DESER) over exactly the same fact; this lesson measured 68,215,021 bytes for MEMORY_ONLY (serialized) over the same data. The difference — more than six times over — comes from a single word: deserialized stores Java objects already ready to use (every string, every double, every timestamp as its own object in memory, with the overhead that implies); serialized stores a compact encoding in raw bytes, which needs decompressing (deserializing) every time a task needs to read a row.

Spark's official documentation summarizes the trade-off precisely:

"If your objects are large, [...] then RDD compaction level should still be sufficient to fit large objects. [...] Don't spill to disk unless the functions that computed your datasets are expensive, or they filter a large amount of the data. [...] Use the serialized RDD storage level if you want fast fault recovery [...] otherwise, recomputing a partition may be as fast as reading it from disk."

Translated into this lesson's practical criterion: deserialized is faster to read (nothing needs decompressing), but takes up more memory; serialized takes up less memory, but every read pays a CPU cost to decompress. Neither is "the right one" in the abstract — it depends on whether your bottleneck is available memory or CPU cycles, a decision you can only make with real numbers like the ones you just measured, not with the default option chosen without thinking.

Common mistakes

Calling .persist(newLevel) on a DataFrame already cached with another level, expecting it to replace it. What happens: someone caches a DataFrame with MEMORY_ONLY, and later in the same script calls .persist(StorageLevel.DISK_ONLY) on the same object, expecting the level to change. Why it happens: it seems reasonable that calling .persist() again, with a different argument, would update the configuration — the same way reassigning a variable would. How to spot it: checking the logs, you're going to find WARN CacheManager: Asked to cache already cached data. — a real warning, verified in this guide, not an error that stops the program. After that warning, df.storageLevel still shows the original level, not the new one: Spark silently ignores the second .persist(). How to fix it: call .unpersist(blocking=True) before calling .persist() again with a different level — exactly the pattern this lesson's worked example used, with an explicit comment flagging it as required, not an extra precaution.

Confusing StorageLevel.NONE's text with a real caching level's text. What happens: someone prints df.storageLevel on a DataFrame that was never cached, sees the text Serialized 1x Replicated, and incorrectly concludes it's already cached, because the text mentions "Serialized" and "Replicated," words that sound like something is being stored. Why it happens: StorageLevel.NONE (the default value for any uncached DataFrame) has a text representation that, due to a quirk of how PySpark builds that text, includes neither the word "None" nor any explicit indicator that no cache is active. How to spot it: if your only evidence that something is cached is .storageLevel's text, with no earlier call to .cache() or .persist() anywhere in the code, be suspicious — also check the object's useMemory/useDisk (df.storageLevel.useMemory, df.storageLevel.useDisk — both False for StorageLevel.NONE, no matter what the text says), or use spark.catalog.isCached(name) on a registered view, as lesson 5 did. How to fix it: never rely solely on .storageLevel's text to decide whether something is cached — combine it with useMemory/useDisk, or with isCached().

Choosing MEMORY_ONLY without considering what happens if the data doesn't fit in memory. What happens: someone chooses MEMORY_ONLY because it sounds like the fastest option, without checking whether the complete DataFrame fits in the cluster's available memory. Why it happens: "memory only" intuitively sounds like the best-performing option, without considering the case where there isn't enough memory for everything. How to spot it: if your DataFrame is larger than available memory and you use MEMORY_ONLY, the partitions that don't fit simply don't get cached — they get recomputed from scratch every time they're needed, instead of failing with an error — which can be surprisingly worse than not caching at all, because part of the work gets cached and part doesn't, with hard-to-predict behavior. How to fix it: if you're not sure the complete DataFrame fits in memory, MEMORY_AND_DISK (.cache()'s base level, without the deserialized variant) is the safer choice — it automatically uses disk for whatever doesn't fit in memory, instead of recomputing those parts every time.

Exercises

Exercise 1 — Confirm the CacheManager warning with your own evidence. Cache a small DataFrame (spark.range(1000), for example) with StorageLevel.MEMORY_ONLY, materialize it with .count(), and without calling .unpersist(), try .persist(StorageLevel.DISK_ONLY) on the same object. Confirm .storageLevel still shows the original level.

See solution 
from pyspark import StorageLevel

df = spark.range(1000)
df.persist(StorageLevel.MEMORY_ONLY)
df.count()
print("level 1:", df.storageLevel)

df.persist(StorageLevel.DISK_ONLY)  # no unpersist() first
print("level 2 (should still be the same):", df.storageLevel)

Expected output (executed in this run, with the real warning in the log):

26/08/13 ... WARN CacheManager: Asked to cache already cached data.
level 1: Memory Serialized 1x Replicated
level 2 (should still be the same): Memory Serialized 1x Replicated

Confirmed: the second .persist() had no effect — the level stayed at Memory Serialized 1x Replicated, the same as the first call, exactly as the log message warns.

Exercise 2 — Calculate the exact ratio between fact's deserialized and serialized sizes. Using the two numbers already measured in this guide — 433,833,928 bytes (deserialized, lesson 5) and 68,215,021 bytes (serialized, this lesson) — calculate how many times heavier the first is than the second.

See solution 
deserialized_bytes = 433_833_928
serialized_bytes = 68_215_021

ratio = deserialized_bytes / serialized_bytes
print(f"the deserialized version weighs {ratio:.2f}x more than the serialized one")

Expected output:

the deserialized version weighs 6.36x more than the serialized one

More than six times over — a contrast large enough to justify, in a real case where available memory is limited, sacrificing some read speed (the cost of deserializing on every access) in exchange for needing much less memory for the same data.

Exercise 3 — Explain, without code, why MEMORY_AND_DISK and MEMORY_ONLY reported the same memSize in this lesson's worked example. In 2-3 sentences, explain why MEMORY_AND_DISK having the option to use disk doesn't mean it's always going to use it.

See solution

MEMORY_AND_DISK stores as many partitions as fit in available memory, and only falls back to disk for the partitions that don't fit — it's conditional behavior, not a guarantee that some disk always gets used. In this lesson's worked example, fact's twelve partitions (68,215,021 bytes total, serialized) fit completely in the driver's available memory, so MEMORY_AND_DISK behaved exactly like MEMORY_ONLY — zero bytes on disk, everything in memory. The difference between the two levels would only become visible with a DataFrame larger than available memory, where MEMORY_ONLY would leave some partitions uncached and MEMORY_AND_DISK would store them on disk instead.

Summary and next step

This lesson measured, with fact_orders_at_scale's same ten million rows, three different StorageLevels: MEMORY_ONLY and MEMORY_AND_DISK (both 68,215,021 bytes, complete in memory in this run) and DISK_ONLY (the same 68,215,021 bytes, all on disk). You also confirmed, with real evidence, why .cache()'s default value (MEMORY_AND_DISK_DESER, measured in lesson 5 at 433,833,928 bytes) weighs more than six times as much as the serialized version — the real cost of storing objects already ready to use instead of a compact encoding.

Before moving on you should be able to: explain the difference between "serialized" and "deserialized" in terms of memory used and read cost; explain why .persist() with a new level has no effect on an already-cached DataFrame, without .unpersist() first; and choose, with real criteria, between MEMORY_ONLY, MEMORY_AND_DISK, and DISK_ONLY based on how much memory you have available.

Lesson 7 completes this module's caching criterion with the opposite case from lesson 5: a cached DataFrame used only once — the same memory cost you just measured, with no benefit justifying it.

Resources

← PreviousWhen caching actually helpsNext →When caching only costs memory