Job, Stage, Task: Decomposing a Run You Have Never Seen
Rebuild the job to stage to task hierarchy of an unfamiliar run from its metrics alone, then name where the time actually went.
The whole decomposition, in three sentences
An action submits a job. The planner cuts that job into stages, one cut at every shuffle. Each stage runs one task per partition, and those tasks run in parallel across the executors' core slots.
Everything else follows from those three sentences. A job with three stages had two shuffles in its plan. A stage with 200 tasks was reading 200 partitions. A stage cannot start until every task of the stage feeding it has finished, which is why one slow task can hold up an entire cluster.
- Actionwrite() submits one job
- Stage 2: scan events6 tasks, writes 640 MB of shuffle
- Stage 3: scan users2 tasks, writes 25 MB of shuffle
- Stage 4: join + write6 tasks, reads 665 MB back, 210 s
- Job done265 s wall clock
Shuffle write belongs to the stage that produced it
Level 6 stated the attribution rule; here it becomes a diagnostic. A shuffle has a write side and a read side, and they sit in different stages. The stage upstream of the boundary writes the shuffle files. The stage downstream reads them back. So a stage whose own transformations are all narrow can still show hundreds of megabytes of shuffle_write, purely because it sits at the boundary and produced the data the next stage needs.
Read it in pairs. Stage 2 writes 640 MB, stage 3 writes 25 MB, stage 4 reads 665 MB: those three numbers are one shuffle seen from both sides. When they do not reconcile, something is missing from your picture, and that is a real diagnostic, not a curiosity.
The walk an interviewer is listening for
Given an unfamiliar slow run, there is a standard order to work in, and following it out loud is most of the answer:
- Which job. Total the runtime per job and find the one that owns the time. Do not tune a 12-second job.
- Which stage. Inside that job, find the stage with the largest share of the job's duration. That is where the fix has to be.
- Which task. Inside that stage, look at how the task durations are distributed. Five tasks around 22 seconds and one at 198 seconds is not a slow stage, it is a skewed one, and the fix is completely different from "add more executors".
Step 3 is where the next module lives. This lesson gets you through steps 1 and 2, which is the part that has to be automatic.
Common mistake: expecting a job's duration_s to equal the sum of its stage durations. It never does. Stages of the same job can overlap when neither depends on the other (the two scan stages of a join run at the same time), and the job also carries scheduling and commit overhead outside any stage. Treat the job duration as the denominator when you compute a stage's share, and never as a total to reconcile.
Interview nuance: "walk me through what happens when you run df.groupBy("x").count()" is the Amazon and TikTok phrasing of this whole lesson. The answer they want is the chain: the count() is the action so it submits one job, the groupBy is wide so the planner cuts the job at that shuffle into two stages, stage one scans and pre-aggregates locally and writes shuffle files, stage two reads them back keyed by x and finishes the aggregation, and each stage runs one task per partition. Saying "it counts the rows in each group" answers a different question than the one asked.
On a real platform this differs. These three tables are the Spark UI's Jobs, Stages and Tasks tabs, trimmed to the columns that carry the lesson. The real Stages tab adds input and output bytes, spill to memory and disk, GC time, and a task-duration percentile summary; the real Tasks tab has roughly thirty columns. On EMR the same data survives the cluster in the persistent application UI, and on AWS Glue each job run links to its own Spark UI archived to S3. The event log behind all of it is newline-delimited JSON, which means teams commonly load a directory of event logs into a warehouse and run exactly the queries you are about to write across every job the platform has ever run.
CREATE TABLE spark_jobs (
job_id INTEGER,
action TEXT, -- the action call that submitted the job
duration_s INTEGER -- wall clock for the whole job, always more than its stages add up to
);
INSERT INTO spark_jobs (job_id, action, duration_s) VALUES
(0, 'count', 25),
(1, 'write', 265),
(2, 'write', 130);
CREATE TABLE spark_stages (
stage_id INTEGER,
job_id INTEGER,
name TEXT,
num_tasks INTEGER,
shuffle_read_mb INTEGER, -- bytes this stage pulled from the previous stage's shuffle files
shuffle_write_mb INTEGER, -- bytes this stage wrote for the NEXT stage to read
duration_s INTEGER
);
INSERT INTO spark_stages (stage_id, job_id, name, num_tasks, shuffle_read_mb, shuffle_write_mb, duration_s) VALUES
(0, 0, 'Scan parquet events + partial HashAggregate', 4, 0, 120, 18),
(1, 0, 'HashAggregate (final)', 4, 120, 0, 6),
(2, 1, 'Scan parquet events', 6, 0, 640, 40),
(3, 1, 'Scan parquet users', 2, 0, 25, 8),
(4, 1, 'SortMergeJoin + write parquet', 6, 665, 0, 210),
(5, 2, 'HashAggregate (final) + write parquet', 4, 900, 0, 120);
CREATE TABLE spark_tasks (
task_id INTEGER,
stage_id INTEGER,
duration_ms INTEGER,
input_records INTEGER,
shuffle_write_mb INTEGER -- this one task's share of its stage's shuffle write
);
INSERT INTO spark_tasks (task_id, stage_id, duration_ms, input_records, shuffle_write_mb) VALUES
( 1, 0, 4400, 2500000, 30),
( 2, 0, 4600, 2510000, 30),
( 3, 0, 4300, 2480000, 30),
( 4, 0, 4700, 2520000, 30),
( 5, 1, 1400, 800, 0),
( 6, 1, 1500, 810, 0),
( 7, 1, 1550, 795, 0),
( 8, 1, 1450, 805, 0),
( 9, 2, 6200, 3000000, 105),
(10, 2, 6400, 3050000, 108),
(11, 2, 6100, 2980000, 104),
(12, 2, 6600, 3100000, 112),
(13, 2, 6300, 3020000, 106),
(14, 2, 6500, 3060000, 105),
(15, 3, 1800, 120000, 12),
(16, 3, 1900, 118000, 13),
(17, 4, 22000, 4100000, 0),
(18, 4, 23500, 4150000, 0),
(19, 4, 21800, 4050000, 0),
(20, 4, 24100, 4200000, 0),
(21, 4, 22600, 4120000, 0),
(22, 4, 198000, 41000000, 0),
(23, 5, 28000, 5000000, 0),
(24, 5, 29500, 5100000, 0),
(25, 5, 27400, 4950000, 0),
(26, 5, 30100, 5200000, 0);-- The hierarchy reassembled: tasks join up to their stage, stages join up to their job.
-- A task row carries no job_id, so the only way to reach the job is through spark_stages.
-- The stage's own shuffle_write_mb and the sum of its tasks' shuffle writes are the same number
-- seen from both sides, which is what makes these metrics worth trusting.
SELECT j.job_id,
j.action,
s.stage_id,
s.name,
COUNT(t.task_id) AS tasks_counted,
SUM(t.input_records) AS input_records,
s.shuffle_write_mb AS stage_shuffle_write_mb,
SUM(t.shuffle_write_mb) AS task_shuffle_write_mb
FROM spark_jobs j
JOIN spark_stages s ON s.job_id = j.job_id
JOIN spark_tasks t ON t.stage_id = s.stage_id
GROUP BY j.job_id, j.action, s.stage_id, s.name, s.shuffle_write_mb
ORDER BY j.job_id, s.stage_id;Apply
Your turn
The task this lesson builds to.
Write a query that returns each job with its stage count, its total task count, the shuffle megabytes its stages wrote, and the input records its tasks read, as (job_id, stages, tasks, shuffle_write_mb, input_records), in job order, over spark_jobs(job_id, action, duration_s), spark_stages(stage_id, job_id, name, num_tasks, shuffle_read_mb, shuffle_write_mb, duration_s) and spark_tasks(task_id, stage_id, duration_ms, input_records, shuffle_write_mb).
A task row carries no job_id, so both task columns have to travel up through spark_stages to reach the job they belong to. Roll the stages up and the tasks up separately before joining them to spark_jobs, or a single flat join will repeat each stage row once per task it ran and inflate shuffle_write_mb.
3 hints and 1 automated check are waiting in the workspace.
Practice
Make it stick
A second problem on the same idea, plus 2 bonus drills.
Write a query that returns the stage that dominates each job's runtime and the share of the job it took, as (job_id, stage_id, stage_share_pct), one row per job, in job order, over spark_stages and spark_jobs.
The dominant stage is the one with the largest duration_s in its job. stage_share_pct is that stage's duration_s as a percentage of the job's own duration_s, rounded to 1 decimal. Use a window function or a correlated subquery, whichever you find clearer.
2 hints and 1 automated check are waiting in the workspace.