The DAG as a Graph
Each DataFrame is a node with parents
A transformation returns a new DataFrame that stores 2 things: the operation, and a link to the DataFrame it was built from. No rows move. Chain 5 transformations and you hold 5 linked nodes, the last pointing back through the others to the source table.
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:
in_stock = products.filter(F.col("in_stock") == 1)
by_category = in_stock.groupBy("category").agg(F.count(F.lit(1)).alias("n"))
cheap = in_stock.filter(F.col("price") < 20)
both = by_category.select("category").union(cheap.select("category"))
both.collect()- products (source, table read)
- in_stock (transform, filter)
- by_category (transform, groupBy + count)
- cheap (transform, filter)
- both (transform, select + union)
- collect() (consumer, action)
- products -> in_stock
- in_stock -> by_category
- in_stock -> cheap
- by_category -> both
- cheap -> both
- both -> collect()
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.
A shared node is shared only in your code
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.
Where the graph is cut into stages
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.
(products
.filter(F.col("in_stock") == 1)
.groupBy("category")
.agg(F.count(F.lit(1)).alias("n"))
.collect())| product_id | category | in_stock |
|---|---|---|
| 1 | Kitchen | 1 |
| 2 | Garden | 1 |
| 3 | Kitchen | 0 |
| 4 | Toys | 1 |
| 5 | Kitchen | 1 |
| 6 | Garden | 0 |
| Stage | Work inside it | Tasks | How it ends |
|---|---|---|---|
| 1 | read products, keep in_stock = 1, start the count per category | 2, 1 per input partition | a shuffle that sends each category to a single task |
| 2 | finish the count for each category | 2, 1 per shuffle partition | 3 rows returned to the driver |
Four of the 6 rows pass the filter (product ids 1, 2, 4 and 5), and the driver receives Kitchen 2, Garden 1 and Toys 1. No stage-2 task can start until both stage-1 tasks finish, because either of them may hold Kitchen rows.
The same graph in the Spark UI
The Spark UI's Jobs page lists each job beside the code line whose action launched it. Open a job and its DAG visualization draws the picture from this section: a box per stage with its operations inside, and an arrow wherever a shuffle connects 2 stages. Count the arrows to count the shuffles. Count how often 1 table's scan appears to catch a branch that reads its source twice.
In the code from this section, how many times does Spark read products when both.collect() runs?
in_stock = products.filter(F.col("in_stock") == 1)
by_category = in_stock.groupBy("category").agg(F.count(F.lit(1)).alias("n"))
cheap = in_stock.filter(F.col("price") < 20)
both = by_category.select("category").union(cheap.select("category"))
both.collect()- Once, because in_stock is 1 DataFrame and both branches share its rows
- Zero, because defining in_stock already read products and the action reuses it
- Twice, because each branch traces back to products separately and the plan holds 2 scans
- Three times: once for in_stock, and once more for each branch built on it
About this lesson
This section is part of the Lazy Evaluation: Intermediate 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.