When caching actually helps

Description

Up to this lesson, every query in this guide read its data source from scratch, every time. That never mattered because, until now, every DataFrame got used in a single query before the script ended. But a real pipeline almost never works that way: the same DataFrame — fact_orders_at_scale, for example — usually feeds several different reports in the same session: an overall total, a per-store breakdown, a running total by date. With no intervention, Spark recomputes everything from the original file on every single one of those queries — the lazy evaluation you already know from module 2 remembers nothing between one action and the next. .cache() changes that: it saves the already-computed result in memory (and, if needed, on disk), so the second and third query reuse it instead of repeating the work. This lesson demonstrates it with three different sources of evidence, none based on time: spark.catalog.isCached(), the .explain() plan, and the bytes read reported by the Spark UI itself.

Connection to the module. Lessons 2 through 4 explained how Spark decides what to execute. This lesson, and the next two, resolve a different question: when it's worth saving a result so you don't have to produce it again.

An analogy: the already-cooked meal, served three times

Pick back up this module's introduction's refrigerator. Imagine you cook a big stew just once, and store it. Throughout the day, three different people open the door hungry and serve themselves from the same stew — none of the three had to cook again, because the work of chopping, seasoning, and cooking was already done the first time. That's, precisely, what this lesson is going to demonstrate: a cached DataFrame reused by three different queries pays the cost of "cooking" — reading the file, computing revenue — once, not three times.

Worked example, part 1: .cache(), isCached(), and the exact moment it materializes

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

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").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"))
fact.createOrReplaceTempView("fact_orders_at_scale")

print("isCached BEFORE .cache():", spark.catalog.isCached("fact_orders_at_scale"))

fact.cache()
print("isCached AFTER .cache(), BEFORE any action:", spark.catalog.isCached("fact_orders_at_scale"))

total = fact.select(F.round(F.sum("revenue"), 2).alias("total")).collect()[0][0]
print(f"\nQuery 1 -- overall total = {total}")
print("isCached AFTER the first action:", spark.catalog.isCached("fact_orders_at_scale"))

spark.stop()

What to expect. Running python3 caching_helps_part1.py, the output is exactly this (executed in this run):

isCached BEFORE .cache(): False
isCached AFTER .cache(), BEFORE any action: True

Query 1 -- overall total = 26537500.0
isCached AFTER the first action: True

Notice something important, which this lesson's common-mistakes section is going to explain in depth: isCached() already reports True before any real action has run — .cache() on its own computes nothing, it only marks the DataFrame as a candidate to be saved; the real work of reading and saving only happens with the first action (Query 1, in this case).

Worked example, part 2: reading the savings in the execution plan itself

# caching_helps_part2.py -- continuation, same fact object, already cached and materialized
print("=== explain() of a SECOND query, reusing the same cached fact ===")
q2 = fact.groupBy("store_id").agg(F.round(F.sum("revenue"), 2).alias("total_revenue"))
q2.explain()

What to expect (executed in this run, over the same fact from part 1, already materialized):

=== explain() of a SECOND query, reusing the same cached fact ===
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[store_id#2], functions=[sum(revenue#8)])
   +- Exchange hashpartitioning(store_id#2, 200), ENSURE_REQUIREMENTS, [plan_id=69]
      +- HashAggregate(keys=[store_id#2], functions=[partial_sum(revenue#8)])
         +- InMemoryTableScan [store_id#2, revenue#8]
               +- InMemoryRelation [order_id#0, franchise_id#1, store_id#2, product_id#3, quantity#4, unit_price#5, order_ts#6, revenue#8], StorageLevel(disk, memory, deserialized, 1 replicas)
                     +- *(1) Project [order_id#0, franchise_id#1, store_id#2, product_id#3, quantity#4, unit_price#5, order_ts#6, (cast(quantity#4 as double) * unit_price#5) AS revenue#8]
                        +- FileScan csv [order_id#0,franchise_id#1,store_id#2,product_id#3,quantity#4,unit_price#5,order_ts#6] Batched: false, DataFilters: [], Format: CSV, ...

This is the first piece of structural evidence: instead of starting with a FileScan csv followed by a Project recomputing revenue — what you'd see if fact weren't cached — the plan starts with InMemoryTableScan, reading directly from an InMemoryRelation. FileScan and Project still show up in the plan, but now they show up inside the InMemoryRelation's definition — a reference to how that cache memory was built the first time — not as work this second query re-runs. No join, no .withColumn("revenue", ...) gets recomputed: q2 reads directly from the already-materialized result.

Worked example, part 3: counting bytes read with and without caching, using the Spark UI's REST API

The most conclusive evidence isn't in the plan — it's in how many bytes Spark had to read from disk, across three different queries. The pattern: run the same three queries twice — without caching and with caching — leaving the SparkSession alive to query its REST API, just as you already did in module 4, lesson 5.

# jobs_no_cache.py
import time
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType, TimestampType

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").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"))
# NO .cache() -- every query rereads the complete CSV from scratch.

print("APP_ID", spark.sparkContext.applicationId)
t1 = fact.select(F.round(F.sum("revenue"), 2).alias("total")).collect()[0][0]
print("Q1 total =", t1)
t2 = fact.groupBy("store_id").agg(F.round(F.sum("revenue"), 2).alias("total")).collect()
print("Q2 =", t2)
t3 = fact.groupBy("product_id").agg(F.round(F.sum("revenue"), 2).alias("total")).collect()
print("Q3 =", t3)

time.sleep(60)  # keep the SparkSession alive to query its REST API
spark.stop()

Run in the background, and queried with curl while it's still alive (http://localhost:4040/api/v1/applications/<app-id>/stages, the same endpoint from module 4):

What to expect (real JSON from this run, summarized to each stage's relevant fields):

stageId=0  status=COMPLETE  numTasks=12  inputRecords=10000000  inputBytes=602074847   -- Q1: reads the CSV
stageId=1  status=SKIPPED
stageId=2  status=COMPLETE  numTasks=1   inputRecords=0

stageId=3  status=COMPLETE  numTasks=12  inputRecords=10000000  inputBytes=602074847   -- Q2: REREADS the complete CSV
stageId=4  status=SKIPPED
stageId=5  status=COMPLETE  numTasks=1   inputRecords=0

stageId=6  status=COMPLETE  numTasks=12  inputRecords=10000000  inputBytes=602074847   -- Q3: REREADS the complete CSV, again
stageId=7  status=SKIPPED
stageId=8  status=COMPLETE  numTasks=1   inputRecords=0

With no caching, each of the three queries reads 602,074,847 bytes (602 MB, kiosko_orders_at_scale.csv's real size) and 10,000,000 records — the complete file, three times, even though the three queries share exactly the same fact. That's nearly 1.8 GB read from disk to answer three questions about the same data.

Now, the same sequence, with fact.cache() added before the first query:

# jobs_with_cache.py -- identical to the one above, with one new line
fact = orders_at_scale_df.withColumn("revenue", F.col("quantity") * F.col("unit_price"))
fact.cache()  # <-- the only new line
# ... the rest of the script is identical: Q1, Q2, Q3, time.sleep(60)

What to expect (real JSON from this run):

stageId=0  status=COMPLETE  numTasks=12  inputRecords=10000000  inputBytes=602074847   -- Q1: reads the CSV AND materializes the cache
stageId=1  status=COMPLETE  numTasks=12  inputRecords=1007      inputBytes=433833928   -- Q2: reads from the CACHE, not the CSV
stageId=2  SKIPPED
stageId=3  status=COMPLETE  numTasks=1

stageId=4  status=COMPLETE  numTasks=12  inputRecords=1007      inputBytes=433833928   -- Q3: reads from the CACHE, again
stageId=5  SKIPPED
stageId=6  status=COMPLETE  numTasks=1

The difference is exactly what the refrigerator analogy predicts: only the first query reads the original file (602,074,847 bytes) — because, besides computing its own result, that first query is the one that materializes the cache. The second and third queries no longer touch kiosko_orders_at_scale.csv at all: they read 433,833,928 bytes from the cached in-memory representation — a different number from the original CSV's, because .cache()'s default cache stores data already deserialized (real Java objects, not text), not a compressed copy of the original text. The total bytes read from disk, across the three queries, drops from ~1.8 GB (without caching) to 602 MB (with caching) — a single file, read once, no matter how many more queries reuse it afterward.

Diagram: three queries, with and without caching

flowchart TD
    subgraph SinCache["WITHOUT cache() -- every query rereads the complete CSV"]
        A1["Q1: FileScan 602 MB"] --> R1["result 1"]
        A2["Q2: FileScan 602 MB (again)"] --> R2["result 2"]
        A3["Q3: FileScan 602 MB (again)"] --> R3["result 3"]
    end

    subgraph ConCache["WITH cache() -- the CSV gets read ONCE"]
        B1["Q1: FileScan 602 MB\n+ materializes the cache"] --> C["InMemoryRelation\n(433 MB, deserialized)"]
        C --> B2["Q2: InMemoryTableScan"]
        C --> B3["Q3: InMemoryTableScan"]
    end

    style A2 fill:#f96,stroke:#333
    style A3 fill:#f96,stroke:#333
    style B2 fill:#9c6,stroke:#333
    style B3 fill:#9c6,stroke:#333

Going deeper: why job count isn't the right metric, and bytes are

It's tempting to look for the caching benefit by counting jobs — but that count is misleading: with or without caching, every one of this lesson's three queries still triggers its own job (more of them, even, counting the internal jobs Adaptive Query Execution submits to measure statistics). What caching eliminates isn't the job count — it's the work inside each job: how many bytes need reading from disk, and how many times revenue needs recomputing from quantity * unit_price for the same ten million rows. That's why this lesson measured inputBytes/inputRecords per stage, instead of counting jobs — it's the metric that truly reflects "less repeated work," the exact phrase describing caching's benefit in this guide.

It's also worth noting the official documentation on what .cache() does under the hood:

"Persists the DataFrame with the default storage level (MEMORY_AND_DISK_DESER)."

MEMORY_AND_DISK_DESER means: store in memory, with data already deserialized (ready to use with no decompression needed), and use disk as a fallback if it doesn't all fit in memory. This module's lesson 6 develops in depth what every word of that name means, and what other options exist.

Common mistakes

Reading isCached() == True as "the data is already in memory." What happens: someone calls .cache(), checks spark.catalog.isCached(...) right away, sees True, and concludes the complete DataFrame is already saved in memory — without having run any action yet. Why it happens: True/False sounds like a simple binary state, and it's natural to assume it describes whether the data is already physically saved. How to spot it: if your code checks isCached() right after .cache(), with no action in between, and uses that result to decide whether "the cost of materializing the cache has already been paid," you're conflating two different things. How to fix it: isCached() answers "is this DataFrame marked to be cached?", not "has it already been saved?". The real data only materializes with the first action — before that, isCached() == True describes an intention, not a done deal. If you need to confirm the data is already physically in memory, check spark.sparkContext._jsc.sc().getRDDStorageInfo() (this module's lesson 6 uses it in depth) after a real action.

Caching a different DataFrame in every query, thinking they reuse the same cache. What happens: someone writes fact.groupBy("store_id").agg(...).cache() in one query, and fact.groupBy("product_id").agg(...).cache() in another, expecting both to reuse the same cached work. Why it happens: both lines start from the same fact, and it's easy to assume "caching something derived from fact" caches fact itself. How to spot it: if you have several lines with .cache() over different transformations of the same base DataFrame, check whether you're really reusing the same cached object, or whether each one builds its own independent cache, sharing nothing. How to fix it: cache the common DataFrame as early as possible in the pipeline — the way this lesson did with fact, before each query transforms it a different way; caching a specific transformation only benefits queries reusing that exact same transformation, not the others.

Caching without having decided, ahead of time, how many times the result is going to get reused. What happens: someone adds .cache() out of habit to any intermediate DataFrame in a pipeline, without having counted how many later queries are going to touch it. Why it happens: after seeing this lesson's real benefit, it's tempting to generalize the rule to "caching always helps." How to spot it: if you can't name, before writing .cache(), at least two concrete queries that are going to reuse that result, you're probably about to pay the memory cost with no benefit — exactly the scenario this module's lesson 7 develops. How to fix it: before caching, count how many times the result gets reused. This lesson demonstrated the benefit with three real queries; lesson 7 shows, with the same evidence discipline, what happens when that count is one.

Exercises

Exercise 1 — Confirm isCached() goes back to False after .unpersist(). On part 1's already-cached and materialized fact, call fact.unpersist(blocking=True) and confirm with spark.catalog.isCached("fact_orders_at_scale") that the result switches to False.

fact.unpersist(blocking=True)
print("isCached AFTER unpersist():", spark.catalog.isCached("fact_orders_at_scale"))

isCached AFTER unpersist(): False

Exercise 2 — Predict how many bytes a FOURTH query would read, reusing part 3's same cached fact. Without running anything, based on the inputBytes pattern you already saw for Q2 and Q3 (both 433,833,928 bytes, reading from the cache), predict what number you'd expect to see for a Q4 — fact.count(), for example — that also reuses the same already-cached and materialized fact.

Exercise 3 — Explain, without code, why 433,833,928 (bytes read from the cache) is a different number from 602,074,847 (the original CSV's bytes), even though both describe the same number of rows. In 2-3 sentences, explain why a cached DataFrame's in-memory size doesn't have to match the size of the file it came from.

Summary and next step

This lesson demonstrated, with three different sources of evidence — spark.catalog.isCached(), InMemoryTableScan's presence in the plan, and the real bytes read according to the Spark UI's REST API — that .cache() over a DataFrame reused across three queries avoids rereading the original file twice: from ~1.8 GB read from disk (without caching, three complete reads) to 602 MB (with caching, a single read, plus 433 MB read from memory per additional query). You also confirmed an important distinction: isCached() marks the intention to cache, not that the data is already materialized — that only happens with the first real action.

Before moving on you should be able to: explain the difference between isCached() before and after the first action; identify InMemoryTableScan in a plan as evidence a query reuses an existing cache; and explain why this lesson measured bytes read instead of counting jobs.

Lesson 6 goes deeper into .persist() and its different StorageLevels — because MEMORY_AND_DISK_DESER, .cache()'s default value, isn't always the right choice, and this guide demonstrates that with the same 10,000,000 rows, measured under three different configurations.

Resources

• Apache Spark — SQL Performance Tuning, "Caching Data" section (.cache()/persist(), unpersist(), and Spark SQL's default caching behavior). spark.apache.org/docs/latest/sql-performance-tuning.html.
• PySpark — DataFrame.cache (official API reference, including the exact quote about MEMORY_AND_DISK_DESER as the default value). spark.apache.org/docs/latest/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.cache.html.
• Apache Spark — Monitoring and Instrumentation (the Spark UI's REST API, /applications/[app-id]/stages, used in this lesson to measure inputBytes with and without caching — already cited in module 4). spark.apache.org/docs/latest/monitoring.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification: .cache() over a DataFrame reused across three queries, verified with isCached() and the repeated work avoided, never with time.