A Job's Life, End to End

Concepts covered: Spark Execution Model

The program below sums profit per region and collects the result, 3 lines that touch every idea in this lesson. Following it from the keystroke to the result shows the order the pieces fire in, and the order is the same for a 6-row table and a 6-billion-row one.

• 1. The 2 transformations append steps to the driver's plan; no executor is contacted
• 2. collect() is the action: the driver turns the plan into tasks and sends them to the executors it already holds
• 3. Each executor reads its partitions and runs 1 task per partition, summing profit per region locally
• 4. Rows for the same region travel across the network to the task that owns that region
• 5. The receiving tasks add the partial sums into 1 row per region
• 6. Each task's result rows travel to the driver, which hands them back to the program

The animation runs it on 6 orders rows read as 2 partitions of 3 rows, across the regions EU, US and APAC, with the shuffle set to 2 output partitions. The first partition holds orders 1 to 3 and the second orders 4 to 6, so the 2 US rows sit in different partitions.

(orders
   .groupBy("region")
   .agg(F.sum(F.col("profit")).alias("total_profit"))
   .collect())

Two things happened before any row moved. Each of the 2 map tasks first computed a partial sum per region from its own rows, so the partition holding both EU orders sent 1 EU row of 55 instead of 2 rows, alongside a US row of 25; the other partition sent US 35 and APAC 30. Then each task wrote its partial rows to local disk, grouped by destination, and that is the moment the first set of tasks finished. The 2 reduce tasks each fetched their rows across the network and added the partials together, which is where the 2 halves of US, 25 and 35, met to make 60 while EU stayed at 55 and APAC at 30. The driver collected 3 rows. That redistribution of rows by key is a shuffle, and it splits a job into 2 sets of tasks, the ones before it and the ones after, which the next tier of this topic takes apart as stages.

Counting what moved

Before the shuffle, the map tasks turned 6 rows into 4 partial rows. Across the network travelled those 4 rows, and 3 reached the driver, so the cluster moved 7 rows for a 6-row input. Scale the table to 6 billion rows over the same 3 regions and the arithmetic holds: each map task still emits at most 3 rows, so a 3,200-partition read sends at most 9,600 partial rows across the network, a few hundred kilobytes, and 3 rows reach the driver. The same job grouped by order_id instead has nothing to combine on the map side, and all 6 billion rows cross the network. Which column a job groups by decides how much moves, and the code for the 2 jobs differs by 1 word.

Where it shows in the Spark UI

Each action starts at least 1 job on the Jobs tab, and this program's collect() is job 0. Opening it shows the task count, 4 here: 2 map tasks and 2 reduce tasks. The task table on each stage's detail page shows which executor ran every task and how long it took. The Executors tab shows how many executors the application holds, with their totals. A first diagnosis is 3 reads of those pages. The job list says which actions fired, the task count says how many partitions the data was in, and the executor count says what the cluster manager granted.

The widget below asks for the same program on the practice seed's orders table, which carries nulls in region that the filter removes before the grouping. The 2 blanks are the aggregation that produces a total per region and the action that brings every result row back to the driver.

Swap collect for first and the job still runs the full grouping, because no region's total exists until every row for that region has been added up; the action only changes how many result rows travel to the driver afterwards.

Do

• Name which process is busy before touching code: the driver, the executors or the manager's grant
• Check the partition count against the slot count before asking for more executors
• End every notebook cell with the action production will use, so its timing means something
• Read upward from the failing action to find the transformation that caused the error
• Keep memory per core in view whenever spark.executor.cores goes up

Avoid

• Don't collect() a DataFrame whose size has not been estimated against the driver's memory
• Don't read a job's speed from take(20) and expect count() to match it
• Don't assume a transformation ran because its line returned without error
• Don't expect a 100-core cluster to parallelise a 1-partition gzip file

About this lesson

This section is part of the Execution Model: 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.