Lazy Evaluation: Beginner

A transformation line returns a new DataFrame and runs nothing. trips.filter(F.col("status") == "completed") records one step, hands back an object that describes the result, and the driver sends no task to any executor. The recorded chain turns into a job only when an action asks for a result. So when an error's cause sits inside the data, the error stays hidden until a task reads that data, and tasks read data only during the action.

The column fare_text holds strings such as "1250", and one row somewhere in the day's 400 files holds "12.50USD". Line 2 carries the bug, because a cast from that string to an integer cannot succeed. Spark cannot see the bad value at line 2: the footer of each file tells Spark that fare_text exists and is a string, and nothing more. So line 2 records the cast and returns. Line 4 builds the job, and because collect() returns fare_total, every row's fare_cents has to be computed. The executors start reading the 400 files, and 18 minutes in, a task reaches the bad row. With ANSI mode on, the default since Spark 4.0, a cast that can't parse its input raises CAST_INVALID_INPUT instead of returning null. The failed task takes the job down with it, and the traceback ends at per_city.collect(), the line the driver was executing when the failure came back.

The table above follows one pattern. Spark checks what it can learn from metadata on the line itself, and leaves everything that needs a row until a task has that row. A misspelled column fails on its own line with cannot resolve, because the schema arrived with the file listing. A value that breaks a cast, a zero that breaks a division, or a file that vanished between the listing and the job can be found only by a task, and a task runs only inside an action.

The cell below records 2 transformations and ends with count(). Run it and note that the only output is the one number the action returns. The filter ran inside the job, and because count() needs no profit_cents value, Spark never computed one. The DataFrame priced itself never held a row.

completed = orders.filter(F.col("status") == "Completed")
priced = completed.withColumn("profit_cents", F.col("profit") * 100)
priced.count()

Every action follows this rule. count() and collect() each start a job that reads every partition, and so does a write, so whichever of them first runs a broken chain is the line the traceback ends at. A step placed after the broken one gives no protection either: a filter on a cast column computes the cast for each row before it can decide whether to keep that row.

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.

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.

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.

completed = orders.filter(F.col("status") == "Completed")
by_status = completed.groupBy("status").agg(F.sum("profit").alias("total_profit"), F.count("*").alias("order_count"))
by_status.collect()

show() and take() ask for a handful of rows, and Spark exploits that. For a chain of per-row steps, a filter followed by a select, a task can produce a finished row as soon as it reads one. Spark starts a job over 1 partition, and if that partition yields 20 rows the job ends there. A show() on a 400-partition table reads 1 partition in a second, while a count() on the same table reads all 400. If the first partition yields fewer than 20 rows, Spark launches another job, and each new round tries about 4 times as many partitions as it has scanned so far, until it has enough rows.

The shortcut works only when a finished row can exist before the whole chain finishes. After a groupBy, no result row is complete until every source row for that key has been seen, so show() on a grouped DataFrame runs the full grouping and then prints 20 of the results. The same holds for orderBy: the first row of a sorted result is known only after every partition has been read. A quick look is cheap on the chain before the grouping, where it costs 1 partition, and costs the whole job after it.

The Jobs page of the Spark UI lists the jobs each action started, which makes the catalog visible after the fact. Every action starts at least 1 job: a pipeline with 1 write usually shows 1, and a notebook cell with 3 show() calls shows 3 or more, because a show() whose first partition comes up short adds follow-up jobs. A line that appears there cost a job, whatever the code looked like.

Picking the action is picking the cost. A look at 5 rows of a filtered table needs an action that stops early. show(5) and take(5) both do, and either one usually reads 1 partition. The number of rows the filter keeps can come only from an action that runs every partition, and count() does that while sending the driver 1 number.

collect() runs the chain and then sends every result row from every executor to the driver, where the rows become Python objects in a list. The executors hold the data spread across many machines, and the driver is a single process, sized for coordinating tasks rather than for holding data. A collect() on a result that fits on one machine is fine. A collect() on a result that doesn't is the most common way a beginner takes down a driver, and it fails only after every executor has done its work.

Take 100 executors finishing a job that produces 50 million rows of 120 bytes. Across the cluster that is 6GB, spread 60MB per executor, which each of them held comfortably. collect() asks all 100 to send their 60MB to the driver, and the driver guards itself with spark.driver.maxResultSize. Once the results that have arrived pass that limit, Spark aborts the job with a SparkException saying the total size of serialized results is bigger than spark.driver.maxResultSize. The executors' work is thrown away, and the failure comes late, after the job has done nearly all its reading. Raising the limit to let 6GB through moves the failure onto the driver itself: its heap is small by default too, it has to hold the whole result at once as Python objects that take more space than the bytes on the wire, and it dies with an OutOfMemoryError. Whether collect() is safe depends only on the size of the result, however large the source is.

collect() is safe after the chain has already shrunk the data. A groupBy("country").agg(...) over 2 billion events returns at most 1 row per country, a few hundred rows, and collecting them into a Python list to build a report costs nothing. A limit(1000) before collect() caps the list at 1000 rows whatever the source holds. The question to ask before typing it is the row count of the DataFrame on the left of the dot. If you cannot bound that count, write the result instead and read it back from where it landed.

The cell below collects, and it is safe because limit(5) caps the result at 5 rows before they travel. Run it, then change 5 to 50 and note that the driver still receives a bounded list.

completed = orders.filter(F.col("status") == "Completed")
completed.select("order_id", "status", "profit").limit(5).collect()

The filter's output from job 1 was stored nowhere, so job 2 cannot pick it up: both jobs list the same 400 files, and the filter runs over every row twice. Job 1 stays cheap here only because a count needs 1 column. Had completed carried a groupBy as well, both jobs would have moved rows between executors and grouped them, and the second job would have cost as much as the first. In a notebook the pattern grows unnoticed, because every show() while exploring is an action: 5 looks at a grouped DataFrame run the full grouping 5 times.

The cell below returns the count and the total profit of completed orders from 1 action. Written as 2 separate actions, Spark would filter the table twice. As one agg, the chain runs once and the driver receives 1 row holding both numbers.

completed = orders.filter(F.col("status") == "Completed")
completed.agg(F.count("*").alias("order_count"), F.sum("profit").alias("total_profit")).collect()

Folding works whenever the numbers you want come from the same chain. One agg call can hold F.count and F.sum next to F.avg and F.min, and F.max too, each with its own alias. The single collect() at the end returns them as the fields of 1 row.

Every action rerunning the chain is also why a failed task can be retried safely: the description is still there, and running it again gives the same rows. A later lesson builds on that.