Lazy Evaluation: Advanced

A DataFrame is a plan, and every transformation returns a new DataFrame whose plan holds the old one as its child. Inside a loop, labels = new_labels rebinds a Python name to the newer plan. The older plan stays reachable as a subtree, so after 30 rounds the name labels points at a single DataFrame whose plan stacks 30 joins and 30 groupBys on top of the original nodes scan.

Here is the job from the intro. The edges table holds each connection in both directions plus a self-loop for every member, so a member's own label is always on offer. Each round, every member takes the smallest label it is offered. Labels can only fall, so once their sum stops falling, no label has moved and the loop stops.

Read the loop the way Spark does. The single line that touches data is first(), the action at the end of the stopping check. Every other line adds operators to a plan held in driver memory. The table follows what labels refers to as each round begins.

The animation runs the same growth on a loop small enough to watch. It starts from Monday's events, adds another day per round with union, and ends in a single collect().

The loop's 3 rounds launch nothing. When collect() runs, 1 job with 1 stage starts: each day's table brings its 2 partitions along, so the stage runs 8 tasks, and the driver gathers all 12 rows. Every round left its operators, and here its partitions too, in the job that finally ran.

This cell rewrites profit 5 times in a loop over orders, then totals it by status. Run it, change range(1, 6) to range(1, 41), and run it again. The totals change; the loop still finishes before any task starts, and all 40 rewrites wait for the collect() at the bottom.

orders = spark.table("orders")
df = orders.filter("status IS NOT NULL").select("status", "profit")

for round_no in range(1, 6):
    df = df.selectExpr("status", "profit * 1.1 AS profit")

result = (
    df.groupBy("status")
    .sum("profit")
    .withColumnRenamed("sum(profit)", "profit_after_rounds")
)
result.collect()

Two lines decide what the cell computes. The reassignment df = df.selectExpr(...) hands each round the DataFrame the previous round built, so the plan stacks every rewrite; without it, each round would discard its rewrite and start over from the unchanged df. And collect() sits after the loop, so all the rounds run inside the single job it launches.

The stopping check ends in first(), an action, and an action runs a plan from its sources. At round k, the check's plan begins at the nodes scan and holds k joins and k groupBys, so Spark recomputes rounds 1 through k before adding up a single label. The job for round k-1 sent its sum to the driver and kept nothing a later job could start from.

Narrow loops pay less of this bill on executors. A loop of withColumn rewrites like the one in the previous section's cell replays cheaply, because pipelining fuses its steps into a single pass over each row. Its cost lands on the driver, which still plans every step. Loops whose rounds hold joins and groupBys pay on both sides.

df.checkpoint() runs the plan of df, writes the resulting rows as files under the checkpoint directory, and returns a new DataFrame whose plan is a single scan of those files. The returned DataFrame keeps no link to the joins that produced it. Any plan built on it starts from the saved rows, so a later check can reach back no further than the most recent save.

Two lines carry the change. setCheckpointDir names a directory on storage that the driver and every executor can reach and that outlives any executor; call checkpoint() without it and Spark raises an error saying the checkpoint directory has not been set. The reassignment new_labels = new_labels.checkpoint() is what cuts the lineage. The call leaves the DataFrame it was invoked on unchanged, so dropping its return value pays for the save and carries on with the uncut plan.

checkpoint() is eager by default: the call itself launches the save, one of those jobs that run before your own action. With checkpoint(eager=False), the call returns at once and the files are written when the first action runs on the returned DataFrame. Loops usually keep the default, so the save lands on a known line and the check right after it reads files.

Rounds of work fall as the interval shrinks, but each save writes the whole labels table to shared storage and adds 2 jobs for the driver to schedule. Saving every round cuts 465 rounds of work to 60, at the price of 30 full writes of the table and 60 extra jobs. The right interval is where another save costs about as much as the replay it removes, and you find it by timing a save job and a round on the Jobs tab. Spark's own ALS recommender exposes the same choice as checkpointInterval, which defaults to 10 and takes effect only once a checkpoint directory is set.

df.localCheckpoint() cuts lineage the same way checkpoint() does: the DataFrame it returns has a plan that starts from saved rows. What differs is where those rows live. A local save keeps each partition in the storage of the executor that computed it, in memory with spill to that executor's local disk. A reliable save writes files to the checkpoint directory on shared storage.

Reliable saves outlive the run. With spark.cleaner.referenceTracking.cleanCheckpoints at its default of false, every save's files stay in the checkpoint directory after the job ends, so a nightly job that saves 6 times a night leaves 180 sets of files behind after a month. Give each run its own subdirectory and delete it when the run succeeds.

Retries cannot fix this, because every attempt looks for the same missing block. After spark.task.maxFailures attempts, 4 by default, the stage is aborted and the job fails in round 26 with 25 rounds of progress lost. Recovery means rerunning from round 1.

Accumulators are Spark's shared counters: tasks add to them, and the driver reads the total. Here a validation function counts payments with no amount as the rows pass through map, and 2 actions follow.

Say 1,000 payments have no amount. count() runs validate over every row and the counter reaches 1,000. The write is a second action, so it reads the table and runs validate again, and the driver prints 2,000. Spark's documentation states the rule: an accumulator update made inside an action is applied once per task, and an update made inside a transformation may be applied more than once if tasks or stages are re-executed.

Move the effect out of the transformation and into an action. foreachPartition is an action that hands your function an iterator over a partition's rows, so the function opens a connection per partition, sends what it needs to, and closes it. Accumulator updates made inside it count once per task, retries included.

An action still runs again under retries and speculation, and an external effect doesn't roll back. A task that delivered 400 of its 500 alerts and then failed delivers all 500 again on its retry. So the receiver has to tolerate repeats, and that needs a stable key taken from the row, here alert_id=row.txn_id. Designing writes that land once on such a key is the subject of the fault-tolerance lessons later in the track.