Why Does Nothing Happen Until `.count()` Runs?
A new engineer on your team writes the following PySpark script and is confused by what they see in the Spark UI:
df = spark.read.parquet("s3://bucket/clickstream/")
step1 = df.filter(F.col("event") == "click")
step2 = step1.withColumn("session_min", F.col("ts") / 60)
step3 = step2.groupBy("page_id").agg(F.count("*").alias("clicks"))
step4 = step3.orderBy(F.desc("clicks"))
print("built the pipeline") # prints almost instantly
top_pages = step4.limit(10).collect() # this line takes 4 minutes
They ask: "Why does building four chained operations on a huge S3
dataset finish in milliseconds, but the very last line takes minutes?
Doesn't each .filter() and .groupBy() have to touch the data to
produce its result?"
- Explain precisely what happens (and doesn't happen) at each of the
four
step*lines. - Explain what happens once
.collect()is called, including which Spark components get involved that weren't involved before. - If they instead wanted to see intermediate progress after each step for debugging, what would you tell them to do, and what would it cost them?
1. What happens at each step* line
Nothing touches data at any of the four lines. spark.read.parquet(...)
only registers a data source in the logical plan — it does not open or
read a file yet. filter, withColumn, groupBy().agg(), and
orderBy are all transformations: each one returns a new
DataFrame object that is really just an updated, still-unresolved
logical plan. Building this plan is a pure in-memory operation on the
driver — no executor is contacted, no task is scheduled, no bytes are
read from S3 — which is exactly why it finishes in milliseconds
regardless of how large the underlying dataset is. The confusion the
engineer has ("doesn't each filter have to touch the data?") is the
single most common Spark misconception: transformations describe
what to compute, not when.
2. What happens at .collect()
.collect() is an action — the only kind of call that triggers
execution. At that line, Spark: (a) resolves the accumulated logical
plan through Catalyst's analysis phase, checking that event, ts,
and page_id actually exist with compatible types; (b) applies logical
optimizations, e.g., pushing the event == "click" filter down into
the Parquet scan itself so unmatched row groups are skipped and unused
columns are never even read; (c) generates a physical plan, choosing
concrete strategies (a hash aggregate for the groupBy, a sort for
orderBy, and noting a shuffle Exchange is required for the
groupBy since rows for the same page_id are scattered across
partitions); (d) the driver splits this physical plan into stages
(a shuffle boundary separates them) and tasks, and only now schedules
those tasks onto executors; (e) executors read the actual Parquet
files, apply the filter and projection, participate in the shuffle for
the aggregation, and the final sorted top-10 rows are sent back to the
driver, which is what .collect() returns. Every component that was
silent before — the cluster manager's granted executors, the shuffle
service, task scheduling — only activates at this one line, which is
why the elapsed time is concentrated there rather than spread evenly
across the four step* lines.
3. Seeing intermediate progress for debugging
Call an action (most commonly .count() or .show(5)) after each
step they want visibility into, e.g., step1.count(),
step2.show(5). This forces Spark to actually execute the plan up to
that point. The cost is real and worth stating explicitly: each
inserted action is a separate job that reprocesses the upstream plan
from scratch (Spark does not automatically remember step1's result for
reuse in step2 unless it was cached), so adding four debugging actions
to a four-step pipeline can multiply the actual work done, not just
add a small profiling overhead. If they need to inspect several
intermediate steps repeatedly (e.g., while iterating in a notebook),
the better approach is step1.cache() (or .persist()) before
calling .count() on it, so the filtered result is materialized in
executor memory once and reused by the following debugging actions
instead of being recomputed from the raw S3 read each time — remembering
to .unpersist() once done so the cache doesn't linger and consume
executor memory needed by other jobs.
Share this question