Lazy Evaluation and DAG Execution

"They kept handing me transformations and I kept saying 'noted.' Only when they finally asked for an answer did I bother to figure out how to get one. Turns out waiting let me skip half the work."

A Scheduler That Refuses to Hurry

In the previous section we built DataFrames and queried them with Spark SQL, and you may have noticed something curious: a line like df.filter(...).select(...) returns instantly even on a terabyte of data, while a later df.count() takes minutes. That gap is the whole subject of this section. The fast calls did almost nothing; they appended a node to a plan. The slow call was an action, and an action is the only thing that makes Spark execute. Understanding this split, and the graph that sits between writing code and running it, is what lets you read a Spark job's behavior instead of being surprised by it.

Every transformation you chain onto a DataFrame, a filter, a select, a groupBy, a join, produces a new logical node and returns immediately. Spark accumulates these into a directed acyclic graph, a DAG, whose nodes are datasets and whose edges are the transformations that derive one from another. The graph is acyclic because each transformation produces a fresh dataset from existing ones; nothing ever loops back. Nothing in this graph computes a single row until you call an action, an operation that must return a concrete value to your program or write data out: count(), collect(), show(), write.parquet(...). We separate the two categories formally in Section 7.5; here we focus on what the deferral buys and how the graph is turned into work.

1. Lazy Evaluation: Record Now, Run Later Beginner

Lazy evaluation means a transformation returns a description of work, not the work's result. When you write df2 = df.filter(cond), no rows are read and no predicate is tested; df2 is simply a handle that says "the rows of df for which cond holds." Chain ten transformations and you have built a ten-node plan while touching zero data. Only an action forces Spark to walk back through the plan, decide how to execute it, and produce the demanded value. This is the opposite of the eager, line-by-line execution of an ordinary Python script, where every statement runs the moment the interpreter reaches it.

The payoff of waiting is that Spark gets to see the complete computation before committing to any of it. An eager engine that ran each transformation immediately would have to materialize every intermediate dataset, write it somewhere, and read it back for the next step, even when the next step is a filter that throws most of it away. A lazy engine, holding the whole plan, can do something far better: it can rewrite the plan. It can move a filter ahead of a projection so fewer columns are carried, collapse two consecutive maps into one pass, and recognize that an intermediate result is never needed on its own and so never build it. The optimizer that performs these rewrites is Catalyst, which we met through Spark SQL in Section 7.3; laziness is precisely what gives Catalyst a whole plan to optimize rather than one isolated operation at a time.

2. The DAG and Its Two Kinds of Edges Intermediate

The plan Spark accumulates is a DAG of partition-level dependencies, and the single most important property of any edge in that graph is whether it is narrow or wide, the distinction introduced for RDDs in Section 7.2. A dependency is narrow when each output partition draws from exactly one input partition: map, filter, and select are narrow, because a row's fate depends only on that row, and the worker holding input partition $k$ can produce output partition $k$ with no help from any other worker. A dependency is wide when an output partition draws from many input partitions, which is the case for groupBy, reduceByKey, join, and repartition: to gather every record with a given key into one place, records must be redistributed across the cluster by their key. That redistribution is the shuffle, the same all-to-all data movement that powered the reduce phase of MapReduce in Chapter 6.

This narrow-versus-wide split is what gives the DAG its execution structure. A run of narrow dependencies can be executed entirely locally: worker $k$ takes its input partition and applies every narrow transformation in sequence, one row flowing through the whole chain, never waiting on another machine. A wide dependency cannot, because the data each output partition needs is scattered across every input partition and must be moved over the network first. So Spark draws its execution boundaries exactly at the wide dependencies. The segments of purely narrow work between shuffles are called stages, and the rule is simple and exact: a new stage begins at every shuffle.

3. The DAG Scheduler: Stages, Pipelining, and Tasks Intermediate

When an action fires, Spark's DAG scheduler turns the logical plan into a physical one in three moves. First it walks the graph and cuts it into stages at every wide dependency, exactly as in Figure 7.4.1. Second, within each stage, it pipelines the narrow transformations: rather than building the output of the map, then reading it back to run the filter, it composes them into a single function that streams each row through the whole chain in one pass. This fusion is the concrete form of the optimization that laziness enabled in Section 1; the intermediate dataset between map and filter is never materialized because the scheduler can see that nothing else needs it. Third, it schedules the stage as a set of tasks, one task per partition, and ships those tasks to the workers holding the relevant data.

A task, then, is the unit of execution: one stage applied to one partition, run by one core on one worker. A stage with 200 partitions becomes 200 tasks that run in parallel across the cluster, gated only by the number of available cores. Stages themselves run in dependency order, and a downstream stage cannot start until the shuffle feeding it has finished writing its output, because that stage's tasks read shuffled data that does not exist until the upstream tasks complete. This is why a Spark job's wall-clock time is dominated by its shuffles: they are the synchronization points where the whole cluster must rendezvous before the next wave of local work can begin. We measure and tune that cost in Section 7.7, where data skew turns a single overloaded shuffle partition into a straggler that holds up an entire stage.

4. Building and Scheduling a DAG from Scratch Advanced

To make the deferral and the staging concrete, the program below implements the whole idea in pure Python with no Spark installed. A LazyFrame records transformations instead of running them, tagging each as narrow or wide. Calling map or filter just appends a node and returns a new frame; nothing executes. Only the action collect triggers planning: it cuts the recorded plan into stages at every wide (shuffle) operation, prints the resulting stage plan, and then executes, fusing the narrow operations within each stage into a single pass. This is a faithful miniature of the DAG scheduler described in Section 3.

class LazyFrame:
    def __init__(self, source, plan=None):
        self.source = source
        self.plan = plan or []      # list of (op_name, kind, fn)

    def _add(self, name, kind, fn):
        return LazyFrame(self.source, self.plan + [(name, kind, fn)])

    # Narrow transformations: each output partition depends on ONE input partition.
    def map(self, fn, label):
        return self._add(label, "narrow", lambda rows: [fn(r) for r in rows])

    def filter(self, pred, label):
        return self._add(label, "narrow", lambda rows: [r for r in rows if pred(r)])

    # Wide transformation: output partitions depend on ALL input partitions (shuffle).
    def group_by_key(self, label):
        def shuffle(rows):
            out = {}
            for k, v in rows:
                out.setdefault(k, []).append(v)
            return list(out.items())
        return self._add(label, "wide", shuffle)

    # DAG SCHEDULER: cut the plan into stages at wide (shuffle) boundaries.
    def plan_stages(self):
        stages, current = [], []
        for op in self.plan:
            name, kind, _ = op
            if kind == "wide":
                if current:
                    stages.append(current)   # close the narrow pipeline before the shuffle
                stages.append([op])          # the shuffle is its own stage boundary
                current = []
            else:
                current.append(op)           # pipeline narrow ops into the running stage
        if current:
            stages.append(current)
        return stages

    # ACTION: planning AND execution happen ONLY here.
    def collect(self, data):
        stages = self.plan_stages()
        print("ACTION triggered: collect()  ->  planning the DAG\n")
        for i, stage in enumerate(stages):
            kind = "SHUFFLE" if stage[0][1] == "wide" else "fused narrow ops"
            ops = " -> ".join(n for n, _, _ in stage)
            print(f"  Stage {i}: [{kind}]  {ops}")
        print()
        rows = data
        for stage in stages:
            for _, _, fn in stage:          # narrow ops in a stage run as ONE fused pass
                rows = fn(rows)
        return rows


# Build the pipeline lazily. NOTHING runs during these calls.
print("Building pipeline (lazy): no computation yet ...")
words = LazyFrame("lines")
pipe = (words
        .map(lambda line: line.strip().lower(), "normalize")
        .filter(lambda line: len(line) > 0, "drop_empty")
        .map(lambda line: (line, 1), "to_pairs")
        .group_by_key("group_by_word")
        .map(lambda kv: (kv[0], sum(kv[1])), "count")
        .filter(lambda kv: kv[1] >= 2, "keep_frequent"))
print(f"Recorded {len(pipe.plan)} transformations, 0 executed.\n")

data = ["Spark", "spark ", "DAG", "spark", "dag", "", "lazy", "DAG"]
result = pipe.collect(data)
print("Result:", sorted(result))

Building pipeline (lazy): no computation yet ...
Recorded 6 transformations, 0 executed.

ACTION triggered: collect()  ->  planning the DAG

  Stage 0: [fused narrow ops]  normalize -> drop_empty -> to_pairs
  Stage 1: [SHUFFLE]  group_by_word
  Stage 2: [fused narrow ops]  count -> keep_frequent

Result: [('dag', 3), ('spark', 3)]

The structure of Output 7.4.1 is the structure of a real Spark job. The first three narrow operations collapsed into one stage that needs no network; the single wide operation became the lone boundary; the trailing narrow operations formed a final stage. If you had added a second group_by_key, a fourth stage would have appeared, because every shuffle adds exactly one boundary. The same plan, run on a cluster, would dispatch one task per partition within each stage and would block the start of Stage 1 until Stage 0's shuffle output was fully written.

5. Reading the Real Plan with .explain() Intermediate

Spark exposes the very plan our toy engine printed. Calling .explain() on a DataFrame shows the physical plan Catalyst chose, and the keyword Exchange in that output marks a shuffle, which is to say a stage boundary. Reading these plans is the practical skill that this section's theory feeds into: the number of Exchange nodes is the number of shuffles, and the number of shuffles plus one is roughly the number of stages your job will run.

from pyspark.sql import SparkSession, functions as F

spark = SparkSession.builder.appName("dag-demo").getOrCreate()
df = spark.read.parquet("events.parquet")

plan = (df
        .filter(F.col("value") > 0)          # narrow: pipelined, no shuffle
        .groupBy("key")                       # wide: forces an Exchange (shuffle)
        .agg(F.sum("value").alias("total"))   # finishes in the post-shuffle stage
        .filter(F.col("total") >= 2))         # narrow: fused after the aggregation

plan.explain()        # prints the physical plan; look for 'Exchange' = shuffle = stage cut
# An abbreviated physical plan looks like:
#   *(2) Filter (total >= 2)
#   +- *(2) HashAggregate(keys=[key], functions=[sum(value)])   <- post-shuffle stage
#      +- Exchange hashpartitioning(key)                        <- the ONE stage boundary
#         +- *(1) HashAggregate(keys=[key], functions=[partial_sum(value)])
#            +- *(1) Filter (value > 0)                          <- pre-shuffle stage
#               +- FileScan parquet [key, value]

from pyspark.sql import functions as F

result = (df
          .filter(F.col("value") > 0)
          .groupBy("key")
          .agg(F.sum("value").alias("total"))
          .filter(F.col("total") >= 2)
          .collect())            # the ONLY action: everything above runs now, optimized

6. Why the Whole Plan, Not Each Step Intermediate

It is worth stating plainly why deferral wins, because it is the reasoning that recurs whenever a system chooses to plan before it runs. If $T_i$ is the work of transformation $i$ and an eager engine ran each in isolation, its cost would be the sum of every step plus the cost of writing and reading every intermediate dataset between steps. The intermediate-materialization term is the expensive one at scale: writing a multi-terabyte result to storage only to read it straight back into the next operation can dwarf the operations themselves. A lazy engine pays almost none of it, because fused narrow operations stream a row through the entire chain in memory, and intermediate datasets that no action observes are never built. The savings is not a constant factor on the operations; it is the elimination of a whole class of I/O.

The same logic explains why Spark plans stages around shuffles specifically. A shuffle is the one unavoidable materialization: its output must be written to disk so that downstream tasks (and re-executed tasks after a failure) can read the right partitions, a property that ties directly to the fault-tolerant re-execution model of Chapter 6. Everything between two shuffles is fair game for fusion and can avoid the disk entirely. So the engine spends its optimization budget exactly where it pays off: collapse the narrow work into streaming passes, and treat each shuffle as a deliberate, costed boundary rather than an accident of how the code was written. This same instinct, see the whole computation, then place the unavoidable communication deliberately, returns when we plan all-reduce schedules for distributed training in Chapter 15.

We now have the execution model that the rest of the chapter rests on: transformations are recorded lazily into a DAG, an action triggers planning, the scheduler cuts the DAG into stages at every shuffle, fuses the narrow work within each stage, and dispatches one task per partition. What we have deliberately left informal is the exact catalog of which operations are transformations and which are actions, and what each costs. That catalog is the subject of Section 7.5, which names every transformation and action, marks each narrow or wide, and shows how the choice between them decides where your shuffles, and therefore your stages, fall.