Lazy Evaluation: Intermediate

Those links form a graph, and its shape starts to matter once your code branches. Two DataFrames can share a parent, and union gives a single node 2 parents. This chain does both:

When both.collect() runs, Spark starts at the action and follows the links backward to every source it can reach. Each node on one of those paths joins the job, and a node on none of them stays a description. Call cheap.collect() instead and the groupBy never runs, because no link leads from by_category to cheap.

Your code shows in_stock once, with 2 children. The plan Spark executes is a tree, so it copies the shared part into each branch: the by_category side gets its own read and filter of products, and the cheap side gets another. One action, 2 scans of the same table. On a 6-row table that costs nothing. On a 300GB table it doubles the read, and no line of your code shows it.

The cost of a branch point is paid once per path that leaves it and reaches the action. Build 3 DataFrames on in_stock, combine them into 1 result, and products is read 3 times.

The scheduler takes the part of the graph an action needs and cuts it wherever rows must move between tasks, the stage boundary from the execution lessons. Everything between 2 cuts runs inside the same task, 1 partition per task. The animation plays a 6-row version of the by_category branch, with products read as 2 partitions and the count finished in 2 shuffle partitions.

That single line launches 2 jobs. The first reads the first line of the input, which gives the column count and, with header=True, the names. The second reads every row of every file to settle each column's type, because a single value of 3.5 at row 40 million turns an integer column into a double. On a 200GB directory that is a full scan, paid again when your first action reads the same files.

File listing still happens during the read. Spark lists the files on the driver, and when the directory tree holds more paths than spark.sql.sources.parallelPartitionDiscovery.threshold (32 by default), it runs the listing as a job of its own.

A pivot turns the distinct values of 1 column into output columns. Spark cannot name those columns without the values, so this line runs a job:

The job finds the distinct status values, sorts them and collects them to the driver. Past spark.sql.pivotMaxValues (10,000 by default) the line fails. Listing the values yourself skips the job and pins the output schema: a new status that shows up next month adds no surprise column for every downstream reader.

Some actions launch more than 1 job before the result arrives. A sort is the common case: orderBy sends rows to tasks by range of the sort key, and to choose the range boundaries Spark first runs a job that samples the key, then runs the work itself. show() is the other: it runs a job over 1 partition, and if that produces fewer than the 20 rows it wants, it runs another over up to 4 times as many partitions, and again, until it has enough. After a filter that keeps 0.01% of the rows, 1 show() can launch several jobs, each scanning more new partitions than the last.

It fails when the write is called, before any task runs, with an AnalysisException: Cannot overwrite a path that is also being read from. The check compares the target path with every path the plan reads. Nothing is deleted.

Calling users.count() before the write would change nothing. An action computes its answer and keeps no rows, the re-execution rule from the beginner lessons, so the write's job goes back to the files regardless.

The guard reads the plan, so it catches only a read the plan still names. Convert the DataFrame to an RDD and back, a common move when the fix is a Python function over rows, and the new plan starts from an anonymous RDD with no path in it. Here clean_row is a Python function that lowercases a row's email:

The RDD still reads /data/users/ lazily, but the plan the guard checks names no file, so the write goes ahead. The old files are deleted, the tasks try to open them, and each fails with an error saying a file does not exist and that the underlying files may have been updated. The job fails, and the path it set out to fix holds no data.

With spark.sql.files.ignoreMissingFiles set to true, which some teams enable to ride out files vanishing during a read, the outcome is worse: the tasks skip the deleted files instead of failing, the job reports success, and the table now has 0 rows.

Sort every transformation by what 1 output partition needs. A narrow transformation builds each output partition from exactly 1 input partition: filter, select, withColumn and union work on the rows in front of them and nothing else. A wide transformation needs rows from many input partitions, because rows that share a key can sit in any of them: groupBy, distinct and orderBy are wide. Every shuffle and every stage cut you have seen so far sits at a wide transformation.

The animation plays a 3-step chain on 6 order_items rows split into 2 partitions: keep rows with quantity >= 2, compute line_total, keep 2 columns.

Run the narrow chain on the practice data. Then add .groupBy("order_id").agg(F.sum("line_total").alias("order_total")) to the end of result and say which steps now run in stage 1 and which in stage 2.

from pyspark.sql import functions as F

result = (order_items
    .filter(F.col("quantity") >= 2)
    .withColumn("line_total", F.col("quantity") * F.col("unit_price"))
    .select("order_id", "product_id", "line_total"))

result.show()

With the groupBy added, stage 1 holds the read, the filter, line_total, the select and the first half of the sum: each task adds up line_total per order_id for the rows it holds. The shuffle sends each order_id to a single task, and stage 2 adds those partial sums into order_total. The 3 narrow steps still cost 1 pass over the input, and the groupBy is the only new boundary in the job.

To place a step you have not met before, ask what 1 output partition needs. A filter on a column, a withColumn with arithmetic and a select need only the row in hand, so they join the current stage. A distinct must compare a row against copies that may sit in any partition, and an orderBy must know where every row ranks, so each one ends the stage. When a wide step cannot be avoided, putting the narrow filters before it means the shuffle moves fewer rows.