🔥 Spark & Databricks · Chapter 3

Transformations vs actions, lazy evaluation

Your filter 'took 0.002 seconds'. It took nothing because it did nothing — yet.

Study this lesson free in the app →

The basics

Lazy evaluation means Spark does not run a transformation when you write it. It records it in a plan and runs the whole plan only when an action asks for a result.

Why it exists: if Spark ran each line immediately, it would read the full rides table, then filter it, then drop columns — materialising huge intermediate results. By waiting until it sees the whole chain, Spark's optimiser (Catalyst) can push the filter into the file scan, read only the three columns you use, and fuse steps into one pass.

Mental model: writing a shopping list versus going to the shop. filter, select, join add lines to the list. count, write, show are the trip. Every trip walks the whole list again, unless you stored the groceries (cache).

Explained

A Meghalaya Cabs engineer wraps each line of a pipeline in a timer. The log says filter: 0.002s, join: 0.001s, write: 41 min.

She spends a day optimising the write. The write was fine. The join was the problem — it just did not run until the write asked for rows.

In Spark, the line that is slow is never the line that is expensive.

The shopping list

You write a list: milk, but only the toned one; eggs; skip anything over ₹200. Writing it costs nothing. Nothing is bought.

When you finally go — that is the action — you read the whole list first, so you can plan a route: dairy aisle once, skip the expensive shelf entirely.

If your partner sends you back for one more item, you walk the whole list again, unless you left the bags in the car. Those bags are cache().

From your code to running tasks

All the cost lands on the action line. That's why timing individual transformation lines tells you nothing.

What laziness buys

When an action fires, Catalyst turns the lineage into a plan in four passes: analysed (resolve columns and types) → optimised logical (rewrite rules) → physical (pick join strategies) → code generation.

The rewrites only possible because nothing ran early:

• Predicate pushdown — the filter moves into the Parquet reader, which skips row groups whose min/max stats exclude COMPLETED
• Column pruning — 3 of 40 columns are read; columnar formats make this an I/O cut of >90%
• Stage fusion — filter, withColumn and the partial aggregation run in one pass per partition, no intermediate DataFrames
• Constant folding, join reordering and choosing broadcast vs sort-merge joins from table sizes

And lineage is the fault-tolerance model: lose a partition, recompute it from the recipe. No replication of intermediate data needed.

Seeing the plan, and timing honestly

from pyspark.sql.functions import col, sum

by_city = (spark.read.parquet('s3://mc-lake/silver/rides/')
    .filter(col('status') == 'COMPLETED')
    .groupBy('city').agg(sum(col('fare_paise') / 100).alias('revenue')))

by_city.explain('formatted')
# FileScan parquet ... PushedFilters: [IsNotNull(status), EqualTo(status,COMPLETED)]
#                      ReadSchema: struct<status:string,fare_paise:bigint,city:string>

# Wrong: times building the plan
t = time.time(); df2 = by_city.filter(col('revenue') > 1e5); print(time.time() - t)

# Right: time an action that forces the work
t = time.time(); by_city.write.mode('overwrite').parquet('s3://mc-lake/gold/rev/')
print(time.time() - t)

PushedFilters and a short ReadSchema are proof that laziness paid off. If your filter is not listed there, something (a UDF, a cast) blocked the pushdown.

At 100×: the double-scan bill

A pipeline that does df.count() for a log line, then df.write, scans a 5TB table twice. At Databricks prices that log line can cost more than the job.

Cache only what is reused and expensive to recompute — a filtered, joined DataFrame feeding three outputs. Do not cache a raw read you use once; you pay memory and gain nothing.

Prefer persist(StorageLevel.MEMORY_AND_DISK) for large reused sets so evicted partitions spill instead of being recomputed from S3. Call unpersist() when done; cached blocks otherwise compete with execution memory.

Very long lineages (iterative loops of hundreds of joins) make planning slow and recovery expensive. checkpoint() or writing an intermediate table cuts the lineage.

Gotchas

• count() for logging — a whole extra job; use metrics from the write, or count once and cache
• Thinking cache() runs immediately — it does nothing until the next action
• Branching without caching — df feeds out_a.write and out_b.write; df's lineage runs twice
• Debugging the wrong line — a UDF that divides by zero fails at write(), with a stack trace pointing at the action
• Python loops of withColumn — each call adds a projection; 300 of them make the plan huge and analysis slow. Use one select with all expressions
• show() as a benchmark — it computes only enough partitions to return 20 rows for simple plans, so it looks faster than the real job

Transformations write the list. Actions go shopping. Every trip walks the whole list.

Time actions, not lines. Cache only what two actions share.

Words to know

In two minutes

• Transformations build a plan and return instantly; actions execute the plan and are the only thing that costs time.
• Each action is a separate job that runs the entire lineage from the source — two actions on an uncached DataFrame read the source twice.
• Laziness lets Catalyst see the whole query: predicate pushdown, column pruning, and fusing narrow steps into one stage.
• cache() is itself lazy: it marks the DataFrame, and the first action fills the cache; later actions reuse it.
• Schema and column errors fail at definition time (eager analysis in classic Spark); data errors — bad casts, UDF exceptions — appear only when an action runs.
• Timing a transformation measures nothing; time actions, and read the plan with explain().

Test yourself

There are 8 questions, 3 puzzles and 5 recall cards for this lesson in the app — the recall cards come back on later days, just before you would forget.