My PySpark Pipeline Was Reading the Same Table 3 Times — Here's Why

My PySpark pipeline was slow. I assumed it needed more cluster resources. It actually needed one line of code.

I had a job that read from a source table, applied a chain of transformations, and then used that same transformed DataFrame in three different places downstream: one write to storage, one aggregation for a summary table, and one validation check. Straightforward pattern. The pipeline worked, but it was taking far longer than the data volume justified.

When I looked at the Spark UI, the pattern became obvious. The source table was being read multiple times, and the exact same chain of transformations was showing up again and again across the job's stages — not once, but three separate times, once for each downstream action that touched the DataFrame.

This post covers why that happens, how to spot it, and the actual fix.

The root cause: transformations are lazy

Spark separates DataFrame operations into two categories: transformations and actions.

Transformations — .filter(), .select(), .join(), .withColumn(), .groupBy() — don't execute anything when you call them. They just build up a lineage: a directed acyclic graph (DAG) describing what would happen if you asked Spark to actually compute the result.

Actions — .count(), .collect(), .show(), .write() — are what actually trigger execution. When you call an action, Spark walks the lineage graph and executes every transformation needed to produce that result.

Here's the part that catches people off guard: every action re-triggers the entire lineage from scratch, all the way back to the original data source, unless you've explicitly told Spark to remember an intermediate result. Spark doesn't automatically cache anything for you between actions.

What this looked like in practice

Here's a simplified version of the pattern I had:

transformed_df = (
    source_df
    .filter(F.col("status") == "active")
    .join(reference_df, "id")
    .withColumn("flag", F.when(F.col("amount") > 1000, 1).otherwise(0))
)

transformed_df.write.parquet("output/main")

summary_df = transformed_df.groupBy("flag").count()
summary_df.show()

invalid_count = transformed_df.filter(F.col("amount") < 0).count()

There are three actions here: .write(), .show(), and .count(). Each one is a separate trigger for Spark to evaluate the lineage of transformed_df. That means:

• The source table gets read from disk three separate times.
• The filter, join, and withColumn transformations each run three separate times.
• Any shuffle involved in the join happens three separate times.

None of this shows up as an error. The job completes successfully. It's just doing two extra passes of work that were entirely avoidable.

Spotting it in the Spark UI

The giveaway is in the Spark UI's SQL / Jobs tab. If you see the same source table scan and the same sequence of transformation stages repeated multiple times across what should logically be one pipeline, that's usually a sign the same DataFrame is being recomputed for each action instead of being reused.

A quick sanity check: count how many actions your pipeline calls on a given DataFrame, and compare that to how many times the source read shows up in the UI. If they match, you've found your redundant computation.

The fix: cache() or persist()

The fix is to tell Spark to materialize and store the result of the transformation chain the first time it's computed, so subsequent actions reuse that stored result instead of recomputing everything.

transformed_df = (
    source_df
    .filter(F.col("status") == "active")
    .join(reference_df, "id")
    .withColumn("flag", F.when(F.col("amount") > 1000, 1).otherwise(0))
    .cache()
)

transformed_df.count()  # first action materializes the cache

transformed_df.write.parquet("output/main")

summary_df = transformed_df.groupBy("flag").count()
summary_df.show()

invalid_count = transformed_df.filter(F.col("amount") < 0).count()

One important detail: .cache() itself is lazy too. Calling it doesn't immediately store anything — it just marks the DataFrame as something to be cached the next time an action runs. That's why I added an explicit .count() right after .cache(), to force the first materialization at a predictable point rather than letting it happen implicitly on whichever action runs first.

After that first action, every subsequent action on transformed_df reads from the cached result instead of re-reading the source table and re-running the transformation chain.

cache() vs. persist()

These aren't two different mechanisms — cache() is a convenience method that calls persist() with a default storage level.

For DataFrames, cache() is equivalent to persist(StorageLevel.MEMORY_AND_DISK). This tries to keep the cached data in executor memory, and spills to disk if it doesn't fit.

persist() lets you choose the storage level explicitly:

from pyspark import StorageLevel

transformed_df.persist(StorageLevel.MEMORY_ONLY)
transformed_df.persist(StorageLevel.MEMORY_AND_DISK)
transformed_df.persist(StorageLevel.DISK_ONLY)
transformed_df.persist(StorageLevel.MEMORY_AND_DISK_SER)

Use MEMORY_ONLY if you're confident the dataset fits comfortably in memory and want the fastest possible reuse. Use MEMORY_AND_DISK (or just .cache()) as a safer default when you're not certain. Use serialized variants (_SER suffix) if memory pressure is a concern and you're willing to trade some CPU time for a smaller memory footprint.

Things I learned that I wish I'd known earlier

Caching isn't automatically a win. Every cached DataFrame occupies executor memory. If you cache too many large DataFrames across a pipeline, you create memory pressure that can slow the job down rather than speed it up — the opposite of the intended effect.

If it doesn't fit in memory, the benefit shrinks. If a cached DataFrame is too large for available memory and has to spill to disk, the performance gain over recomputation can become marginal. It's still often better than recomputing an expensive join or shuffle from scratch, but it's not the dramatic speedup you'd get from an in-memory cache hit.

Don't forget to release it. Calling .unpersist() once you're done with a cached DataFrame frees up that memory for the rest of the job. Skipping this step means the memory stays reserved even after you no longer need the cached data, which can starve other operations running later in the same job.

transformed_df.unpersist()

Cache after your transformation chain is final, not mid-chain. If you cache an intermediate DataFrame and then keep adding more transformations on top of it, you may end up caching a version of the data you don't actually need cached, while the version that's genuinely reused downstream never gets the benefit.

Only cache what's actually reused. If a DataFrame is used in exactly one action, caching it adds overhead (the cost of materializing and storing it) without any payoff, since there's no second computation to avoid.

A simple rule of thumb

Cache a DataFrame when all of the following are true:

• It's referenced by more than one action downstream.
• Recomputing it is expensive — it involves a join, a wide shuffle, or a costly aggregation.
• It comfortably fits within your cluster's available memory, or you're deliberately choosing a disk-backed storage level as a tradeoff.

Skip caching when a DataFrame is used exactly once, when it's cheap to recompute, or when your cluster is already under memory pressure from other cached data.

The broader lesson

Spark's lazy evaluation model is a deliberate design choice — it lets Spark's optimizer look at the full chain of transformations and plan an efficient execution strategy rather than executing operations one at a time as they're written. But that same laziness means the responsibility falls on you to recognize when a DataFrame is about to be computed more than once.

A pipeline that "needs a bigger cluster" is sometimes actually a pipeline doing the same work two or three times over. Throwing more compute at that problem makes the redundant work finish a bit faster — it doesn't eliminate the redundancy. Adding one .cache() call in the right place fixed what more executors never would have.

Have you caught a Spark job silently recomputing the same DataFrame multiple times? What tipped you off — the Spark UI, a runtime that didn't add up, or something else?