Re-Execution on Every Action

A DataFrame is a description of how to compute rows, and after a job finishes, the rows the tasks produced are gone. The driver keeps whatever the action returned, whether a number or a list. The executors keep nothing. So when a second action runs on the same DataFrame, Spark has no rows to reuse, and it runs the chain again from the source. This is the cost of laziness, and it bites every pipeline that calls one action for a check and another for the output.

trips = spark.read.parquet("s3://rides/trips/2024-03-14/")
completed = trips.filter(F.col("status") == "completed")
print(completed.count())
completed.write.parquet("s3://rides/completed/2024-03-14/")

The filter's output from job 1 was stored nowhere, so job 2 cannot pick it up: both jobs list the same 400 files, and the filter runs over every row twice. Job 1 stays cheap here only because a count needs 1 column. Had completed carried a groupBy as well, both jobs would have moved rows between executors and grouped them, and the second job would have cost as much as the first. In a notebook the pattern grows unnoticed, because every show() while exploring is an action: 5 looks at a grouped DataFrame run the full grouping 5 times.

The Jobs page of the Spark UI makes this visible. 2 jobs in a row that read the same source with the same input size are the tell that one DataFrame was acted on twice, and the fix starts from the code that called the second action.

How many times does the program below read the source files under s3://logs/events/?

events = spark.read.parquet("s3://logs/events/")
errors = events.filter(F.col("level") == "ERROR")
print(errors.count())
errors.write.parquet("s3://logs/errors/")
print(errors.count())

• Once, because errors is computed on line 2 and the 3 actions reuse it.
• Twice, because the write stores the filtered rows and the last count reads them.
• Three times, once per action, with the filter running over every row 3 times.
• Four times, once for the read on line 1 and once more for each action.

Why Spark forgets

Spark drops task output on purpose. A job over 200GB produces hundreds of gigabytes of intermediate rows across the executors, and keeping all of it after every job would fill the cluster's memory within a few actions. Throwing it away costs nothing when no second action comes, and most DataFrames in a pipeline are acted on once. The description is cheap to keep and to rerun, and Spark bets on that. When you do need a result twice, you have to tell Spark. The tools for that within reach at this point are below. A later lesson covers asking Spark to hold a result in memory between actions.

3 ways to avoid the second pass

• Fold the actions into one: A count and a sum you want together belong in one agg(F.count("*"), F.sum("fare_cents")) followed by one collect(). The chain runs once and the driver gets both numbers.
• Write once, read the written data: Write completed to storage, then point later actions at a fresh spark.read of that path. Each later action pays for the filtered data, and the filter runs once.
• Take exploratory actions out of the pipeline: A count() left in from debugging adds a pass over the source every night. Drop it, or count a limited sample during development only.

completed = trips.filter(F.col("status") == "completed")
completed.write.parquet("s3://rides/completed/2024-03-14/")
saved = spark.read.parquet("s3://rides/completed/2024-03-14/")
print(saved.count())

The cell below returns the count and the total profit of completed orders from 1 action. Written as 2 separate actions, Spark would filter the table twice. As one agg, the chain runs once and the driver receives 1 row holding both numbers.

Folding works whenever the numbers you want come from the same chain. One agg call can hold F.count and F.sum next to F.avg and F.min, and F.max too, each with its own alias. The single collect() at the end returns them as the fields of 1 row.

Every action rerunning the chain is also why a failed task can be retried safely: the description is still there, and running it again gives the same rows. A later lesson builds on that.

About this lesson

This section is part of the Lazy Evaluation: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.