Caching Decisions, and the Slow-Job Investigation
Decide when caching pays from the evidence in the metrics, then run the whole Amazon slow-job walk on a snapshot you have never seen and check your diagnosis against the model answer.
cache() and persist()
cache() asks Spark to keep a dataset's computed partitions around after the first action, so the next action reads them instead of recomputing the whole lineage. persist(storageLevel) is the same request with the storage level spelled out: memory only, memory and disk, serialized, replicated. cache() is exactly persist(MEMORY_AND_DISK) on a DataFrame, so the choice is really about how you want to pay when memory runs short.
When caching pays, and when it costs
Caching pays when a dataset feeds two or more actions. One action, one pass, nothing to reuse, and the cache is pure overhead.
Caching hurts when the cached dataset is large enough that keeping it evicts everything else. Then the partitions get dropped anyway, you recompute them on the next access, and you also slowed down the work that lost its memory. "Cache everything" is not a strategy; it is a way to convert free memory into a slower job.
The tell in the metrics is specific and worth memorizing: the same scan stage appears in more than one job. Every repeat after the first is work you already did. That is the query the Apply exercise grades, and it is the evidence you would show someone before adding a cache() call.
The metric walk
When someone hands you a slow job and no context, the order matters more than any single query. This is the walk:
- Job durationwhich job is actually slow
- Dominant stagewhich stage owns most of that job's time
- Task distributionmax over median inside that stage
- Shuffle volumehow much data that stage moved
- Configpartitions, cores, AQE on or off
Skipping to step three is the most common way to waste an interview: you compute a beautiful skew ratio for a stage that accounts for four percent of the runtime.
The question this lesson is built from
Amazon asks a version of this in its junior data-engineering loop, and the phrasing is close to: your EMR job takes 4 hours, the Spark UI shows one task that ran for 40 minutes against a 2-minute median, diagnose it and tell me what you would change.
The snapshot you are about to query is not that job. It is a smaller capture with the same shape, and the shape is the part that transfers: one stage owning most of the run, one 40-minute straggler inside it against a median task far below it, a scan that repeats across jobs, and a config that explains why nobody split anything. Three jobs, eleven stages, a sampled Tasks-tab export covering the three longest stages, and the cluster config. Read your own numbers off the snapshot rather than reusing the ones in the question. Before you read any further, do this: run the Apply query, then the Practice query, and write down your own diagnosis in three or four sentences. What is wrong, what is the evidence, what would you change first.
Model answer: read this only after you have written your own
A strong candidate answers in this order.
Where the time went. Stage 4, join_orders_customers, is 2,700 seconds out of 4,172 seconds of total stage time, so roughly two thirds of the whole run sits in one stage. Everything else is noise until that stage is explained.
What is wrong with it. The task distribution inside stage 4, across the tasks the export sampled, is 2,400,000 ms max against a 400,000 ms median, a 6x ratio. That is skew, not size: if the stage were merely big, every task would be slow. One task also spilled 6 GB to disk while no other task spilled at all, which corroborates that it was holding more data than its share of memory, not just running on an unlucky machine.
What I would change, in order. First, turn AQE on. spark.sql.adaptive.enabled is false in this snapshot, so nothing was ever going to split anything at runtime, and that is a one-line change. I would say what it depends on rather than promising it: AQE splits when a partition is more than 5 times the median partition SIZE and larger than 256 MB, and this capture carries stage and task metrics but no per-partition sizes, so I would read those two numbers off the shuffle-partition view before calling it the whole fix. Second, if the customer side is small enough, broadcast it: under 10 MB the planner does it unasked, up to about 100 MB a hint will do it, and that removes the shuffle so the hot key stops mattering. Third, if neither applies, salt the hot key on the orders side and replicate the customer side across the salt values.
Two more findings the snapshot holds. scan_orders_parquet runs in all three jobs, at 180, 195, and 186 seconds, so about 381 seconds is spent recomputing a scan that never changed. That dataset feeds three actions, which is exactly the caching criterion. scan_returns_parquet also runs twice, but both runs sit inside job 2, so that is one action reading the same table on two branches of its plan rather than a dataset being recomputed for a second action; caching it buys much less, which is why the recomputation query counts distinct jobs and not stage rows. And spark.sql.shuffle.partitions is 200 on a cluster of 16 executors at 4 cores each, which is 64 cores; the 2-to-4-per-core heuristic wants 128 to 256, so 200 is actually reasonable here, and saying so is better than reciting "200 is always wrong". The real config problem is AQE being off.
Now compare that against what you wrote and mark your own gaps. The gaps are the study list.
Common mistake: caching a dataset that is read once. Read the metrics first, and only reach for cache() when the same stage genuinely appears in more than one job.
Interview nuance: the answer above is roughly 200 words and follows the walk in order. Candidates who lead with the fix ("I would salt it") before the measurement lose the point even when the fix is right, because the interviewer is testing whether you can find a problem you have not seen before, not whether you know the word "salting".
On a real platform this differs. You would be reading this from the Spark History Server, from Glue job metrics in CloudWatch, or from Databricks' query profile, and the numbers arrive as charts rather than tables. The walk is identical: job, then dominant stage, then distribution, then volume, then config. Warehouses give you the same walk under different names, a query profile with per-operator time, bytes spilled, and partitions pruned.
CREATE TABLE spark_jobs (
job_id INTEGER,
action TEXT, -- the action that submitted this job
duration_s INTEGER
);
INSERT INTO spark_jobs (job_id, action, duration_s) VALUES
(1, 'count', 430),
(2, 'write', 3270),
(3, 'collect', 580);
CREATE TABLE spark_stages (
stage_id INTEGER,
job_id INTEGER,
name TEXT,
num_tasks INTEGER,
shuffle_read_mb INTEGER,
shuffle_write_mb INTEGER,
duration_s INTEGER
);
INSERT INTO spark_stages (stage_id, job_id, name, num_tasks, shuffle_read_mb, shuffle_write_mb, duration_s) VALUES
(1, 1, 'scan_orders_parquet', 200, 0, 4096, 180),
(2, 1, 'aggregate_count', 200, 4096, 0, 240),
(3, 2, 'scan_orders_parquet', 200, 0, 4096, 195),
(4, 2, 'join_orders_customers', 200, 4096, 5120, 2700),
(5, 2, 'write_parquet', 48, 5120, 0, 110),
(6, 2, 'filter_active_users', 200, 0, 512, 45),
(7, 3, 'scan_orders_parquet', 200, 0, 4096, 186),
(8, 3, 'aggregate_by_region', 200, 4096, 0, 320),
(9, 3, 'filter_active_users', 200, 0, 512, 52),
(10, 2, 'scan_returns_parquet', 200, 0, 2048, 70),
(11, 2, 'scan_returns_parquet', 200, 0, 2048, 74);
CREATE TABLE spark_tasks (
task_id INTEGER,
stage_id INTEGER, -- a SAMPLE of the Tasks tab: seven of each of the three longest stages'
-- 200 tasks, which is what the UI hands you when you export a page of it
duration_ms INTEGER,
spill_mb INTEGER
);
INSERT INTO spark_tasks (task_id, stage_id, duration_ms, spill_mb) VALUES
(201, 2, 36000, 0),
(202, 2, 39000, 0),
(203, 2, 42000, 0),
(204, 2, 44000, 0),
(205, 2, 47000, 0),
(206, 2, 50000, 0),
(207, 2, 66000, 0),
(401, 4, 250000, 0),
(402, 4, 320000, 0),
(403, 4, 380000, 0),
(404, 4, 400000, 0),
(405, 4, 460000, 0),
(406, 4, 520000, 0),
(407, 4, 2400000, 6144),
(801, 8, 42000, 0),
(802, 8, 45000, 0),
(803, 8, 48000, 0),
(804, 8, 50000, 0),
(805, 8, 52000, 0),
(806, 8, 55000, 0),
(807, 8, 58000, 0);
CREATE TABLE cluster_config (
setting TEXT,
value TEXT
);
INSERT INTO cluster_config (setting, value) VALUES
('spark.sql.shuffle.partitions', '200'),
('spark.executor.instances', '16'),
('spark.executor.cores', '4'),
('spark.driver.memory', '8g'),
('spark.sql.adaptive.enabled', 'false');-- The same scan stage, once per job. Three runs of identical work, and only the
-- first one was necessary.
SELECT j.job_id, j.action, s.stage_id, s.name, s.duration_s
FROM spark_stages s
JOIN spark_jobs j ON j.job_id = s.job_id
WHERE s.name = 'scan_orders_parquet'
ORDER BY j.job_id;Apply
Your turn
The task this lesson builds to.
Write a query that returns every stage name that ran in more than one job, with how many times it ran and the seconds wasted beyond its fastest run, as (name, times_run, wasted_s), most wasted first, over spark_stages(stage_id, job_id, name, num_tasks, shuffle_read_mb, shuffle_write_mb, duration_s).
A stage name counts only if it appears in more than one distinct job_id. times_run is how many stage rows carry that name. wasted_s is the name's total duration_s minus its single fastest run, which is the time caching would have saved.
3 hints and 1 automated check are waiting in the workspace.
Practice
Make it stick
A second problem on the same idea, so it survives past today.
Write a query that returns the snapshot's dominant stage with what it costs its own job, as (stage_id, name, share_of_job_pct, p75_ms), over the snapshot's spark_stages and spark_tasks tables.
The dominant stage is the one with the largest duration_s. share_of_job_pct is that stage's duration_s as a percent of the summed duration_s of every stage carrying the same job_id, rounded to 2 decimals, which is the number that says whether fixing this stage fixes the job it sits in. p75_ms is the same stage's 75th-percentile task duration by nearest rank: order its sampled tasks by duration and take the row at position (3 * n + 3) / 4 under integer division. One row out.
1 automated check is waiting in the workspace.