Why Laziness Makes Spark Fast
Concepts covered: Catalyst Optimizer
Lazy evaluation looks like a quirk until you see what it buys. Because no step runs until the action, Spark holds the entire chain, from the read to the action, before it touches a byte. A library that runs each line as it arrives knows only that one line, so it has to produce the full result of line 1 before it learns what line 2 wants. Spark knows what every later line wants, and it uses that knowledge to skip work the chain never needs.
What the whole chain lets Spark skip
trips = spark.read.parquet("s3://rides/trips/2024-03-14/")
completed = trips.filter(F.col("status") == "completed")
slim = completed.select("city", "fare_cents")
per_city = slim.groupBy("city").agg(F.sum("fare_cents").alias("fare_total"))
per_city.collect()Take the day's trips as 200GB across 40 columns of roughly equal size, 5GB per column, with 7 trips in 10 completed. Run eagerly, line by line, line 1 loads all 200GB, line 2 writes a 140GB copy holding the completed rows, line 3 writes a 7GB copy holding 2 columns of that, and only then does the grouping begin over the 7GB copy. Spark runs the same 5 lines as one job. Apart from status, the chain mentions only city and fare_cents, and collect() needs fare_total, so the executors read those 3 columns, 15GB. Each task filters its rows as they stream through, and the 3 in 10 that fail the filter are dropped before anything is sent anywhere. The grouping needs every city's total in one place, and each task adds up its own rows per city before anything moves, so what travels between executors is one small row per city per task, a few MB in all, rather than 7GB of trips.
| Step | Eager, line by line | Spark, whole chain |
|---|---|---|
| Read | 200GB, all 40 columns | 15GB, the 3 columns the chain names |
| Filter | Writes a 140GB copy | Drops rows inside each task, writes nothing |
| Select 2 columns | Writes a 7GB copy | Already done at the read |
| Group by city | Reads the 7GB copy back and groups it | Sums per city inside each task, moves a few MB |
| Total data handled | Over 350GB | About 15GB |
The table gives 3 rules, and each one depends on Spark seeing the chain before it runs it.
- Read only what the chain names: Spark collects every column any line uses and asks the source for those alone. A format that stores each column separately hands back 3 columns and never touches the other 37.
- Filter before anything moves: A filter runs inside the task that read the row, so a dropped row costs one comparison and never crosses the network or lands on disk.
- One pass, no copies in between: A filter followed by a select and then a withColumn executes as one pass over each partition, as does any run of per-row steps. A row goes in one end and out the other, and the DataFrames in between are names in your code, never data on a machine.
In the 5-line trips program above, line 2 assigns the DataFrame completed. How many rows does completed hold anywhere in the cluster right after line 2 runs?
- Seven in 10 of the source rows, the completed ones, spread across the executors.
- All of the source rows, because the filter runs when the next line asks for columns.
- The rows of the first partition, which Spark reads to check that the filter compiles.
- None: completed describes a filter and holds no rows until an action runs the chain.
Watching the chain run
The animation below runs a filter and a groupBy over 8 orders split into 2 partitions, ending in a collect(). Step through it and watch the timeline: the first 2 lines produce no task at all. When collect() arrives, 2 tasks read the 2 partitions and drop orders 3 and 5 inside the task that read them. Each task then adds up its own completed orders per region and sends those partial totals on. The first partition's totals are US 120 and EU 80, plus 60 for APAC. The second's are US 95 and EU 55, plus 70 for LATAM. The 2 grouping tasks combine them into 4 regional totals: US reaches 215 and EU 135, while APAC stays at 60 and LATAM at 70. The driver receives those 4 rows. Orders 3 and 5 never travelled, and no table of 6 filtered rows was ever stored between the 2 lines.
completed = orders.filter(F.col("status") == "Completed")
by_region = completed.groupBy("region").agg(F.sum("profit").alias("total_profit"))
by_region.collect()| order_id | region | status | profit |
|---|---|---|---|
| 1 | US | Completed | 120 |
| 2 | EU | Completed | 80 |
| 3 | US | Pending | 45 |
| 4 | APAC | Completed | 60 |
| 5 | EU | Cancelled | 30 |
| 6 | US | Completed | 95 |
| 7 | LATAM | Completed | 70 |
| 8 | EU | Completed | 55 |
Run the cell below. The chain filters and groups by status before it collects, all in one job. The filter removes every order whose status reads anything other than Completed before the grouping sees a row, which is why the result holds 1 row.
About this lesson
This section is part of the Lazy Evaluation: Beginner lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.