Skip to main content

Module 3 — Transformations, actions and lazy evaluation

The DataFrame API of the previous module has a property that is easy to state and easy to forget in practice: nothing runs until you ask for a result. Every filter, select or groupBy returns a new DataFrame that only records what would be done. This lets Catalyst rewrite the whole pipeline as a single plan — and it explains most of the surprises beginners hit.

Transformations are recipes, actions are triggers​

A transformation is any operation that returns another DataFrame: filter, select, withColumn, join, groupBy followed by agg, orderBy. None of them touch the executors on their own.

An action is an operation that returns a value or writes something: count(), collect(), show(), first(), write.parquet(...). These trigger the entire chain that produced them.

flights = spark.read.parquet("s3://open-data/flights/")

step1 = flights.filter(F.col("cancelled") == 0)           # nothing happens
step2 = step1.withColumn("late", F.col("arrival_delay") > 15)  # still nothing
step3 = step2.groupBy("carrier").agg(F.avg("late").alias("rate"))  # still nothing

step3.show()                                              # everything runs now

Running the first three lines feels instant, and rightly so — Spark only recorded the plan. Running .show() at the end causes it to read the entire dataset, apply every transformation, and produce five rows. Understanding this timing is the difference between a job that runs in three minutes and a job that runs the same code three times because you called .count() between every step.

Narrow and wide, and why the shuffle matters​

Transformations fall into two families that behave very differently at scale.

A narrow transformation is one where each output partition depends on at most one input partition. filter, map, withColumn on an existing column and select are narrow. They cost roughly what reading the data costs; nothing moves across the network.

A wide transformation is one where each output partition depends on many input partitions. groupBy followed by an aggregation, join, orderBy and distinct are wide. Executing them requires a shuffle: intermediate data is written to disk on every executor, then pulled by other executors according to a hash of the grouping key.

The shuffle is the single most expensive operation Spark can perform. It writes the entire intermediate dataset to disk, transfers a large fraction of it across the network, and creates a stage boundary. On the flight-delays project a single well-placed shuffle takes tens of seconds; three carelessly stacked shuffles turn a two-minute job into a twenty-minute one.

Reading the plan​

explain() prints the plan Catalyst produced. Reading it is the fastest way to spot an unnecessary shuffle.

step3.explain()
# == Physical Plan ==
# AdaptiveSparkPlan
# +- Sort [rate DESC NULLS LAST], true
#    +- Exchange rangepartitioning(rate DESC, 32)   <-- SHUFFLE for the sort
#       +- HashAggregate(keys=[carrier], ...)
#          +- Exchange hashpartitioning(carrier, 32)  <-- SHUFFLE for the group by
#             +- Project [carrier, cast((arrival_delay > 15) as double) AS late]
#                +- Filter (isnotnull(cancelled) AND (cancelled = 0))
#                   +- FileScan parquet [carrier, arrival_delay, cancelled]

Every Exchange line is a shuffle. In the plan above there are two: one to group by carrier, one to sort by delay rate. The FileScan line already shows Catalyst's work — arrival_delay and cancelled are the only columns read, and the cancelled = 0 filter is pushed all the way down to the reader.

Caching is not free, and it is not automatic​

Because transformations are lazy, calling an action twice on a DataFrame recomputes the whole chain twice. This is the second most common Spark surprise: writing df.count() "just to see" adds a full pass over the data.

model_features = flights.filter(...).join(weather, "airport").withColumn(...)

model_features.cache()          # marks it as "keep in memory"
model_features.count()          # first action materializes it

train, test = model_features.randomSplit([0.8, 0.2])   # both reuse the cache

.cache() is a hint to keep the DataFrame in memory after its next materialization. Without a following action, it does nothing. .persist(StorageLevel.MEMORY_AND_DISK) gives you finer control if the cached data would not fit in RAM. Both cost real memory on the executors; only cache what you actually reuse, and unpersist explicitly once you no longer need it.

Every action triggers a full plan

show(), count(), collect(), write — each one relaunches the entire chain unless the intermediate result was cached. Debugging a Spark job by peppering it with .count() calls is the fastest way to make it slower than a Python script.

Summary​

  • Transformations describe work, actions trigger it; nothing happens until an action is called.
  • Narrow transformations stay inside a partition; wide transformations force a shuffle and dominate cost.
  • explain() shows the physical plan; every Exchange line is a shuffle you should be able to justify.
  • cache() and persist() avoid recomputation but cost memory; call an action after them, and always match with unpersist().

Next module: reading the data itself — columnar formats, explicit schemas and partition pruning.