The Driver, the Executors, and One Task per Core
The two kinds of process in every Spark application, how work lands on core slots one task at a time, and the driver-versus-executor OOM fork read straight off the executor metrics.
One driver, many executors
A Spark application is two kinds of process, and almost every Spark question you will be asked resolves to knowing which one you are talking about.
The driver is the single process that runs your main(). It builds the logical plan, hands it to the optimizer, cuts the physical plan into stages, decides which task goes to which machine, and collects results that you explicitly ask to bring back. There is exactly one driver per application and you cannot scale it out. On EMR or Glue it is one JVM on one machine, and if it dies the whole application dies with it.
The executors are the worker processes. Each one is a JVM that holds partitions of your data in its heap, runs the tasks the driver assigns, caches what you tell it to cache, and writes shuffle files for the next stage to read. You scale out by adding executors, and you scale each one up by giving it more cores or more memory.
| the question | driver (exactly one) | executor (as many as you buy) |
|---|---|---|
| runs what? | your main(), the optimizer, the scheduler | the tasks the driver sends it |
| holds what? | the plan, plus anything you collect() back | partitions of the data, cached blocks, shuffle files |
| scales how? | it does not, there is only ever one | add executors, or cores and memory per executor |
| dies how? | collect() or toPandas() on a big result set | too few partitions, or too many cores sharing one heap |
| the knob | spark.driver.memory | spark.executor.memory, spark.executor.cores |
One partition, one task, one core slot
The unit of parallel work is the task, and the rule is mechanical: one partition of the data becomes one task, and one task occupies one core on one executor for its entire duration. So a cluster with four executors of four cores each has 16 slots and can run 16 tasks at once. A stage with 200 partitions runs 200 tasks through those 16 slots in waves, and the stage is not finished until the last wave is.
That arithmetic is why the two numbers on an executor are not independent. Cores decide how many tasks run on it at once. Memory is the heap those tasks share. Doubling the cores on an executor without doubling its memory halves the memory each concurrent task can use, which is not a tuning win, it is a way to convert a working job into a failing one.
The OOM fork
"My Spark job ran out of memory" is not a diagnosis. There are two completely different failures behind it, and the interviewer is checking whether you know to ask which.
- Driver OOM. Something pulled the dataset back to the single driver process.
collect(),toPandas(), or an unboundedtake()on a large result. The fix is not more driver memory, it is not collecting: write the result to storage, or aggregate first and collect the small answer. - Executor OOM. A single task tried to hold more than its share of one executor's heap. The usual causes are too few partitions (so each one is enormous), too many cores per executor sharing the same heap, or a skewed key that dumps a huge group into one task. The fix is more partitions, fewer cores per executor, or dealing with the skew.
The metrics tell you which. If the driver process died and the last thing in the log was a collect(), that is the first branch. If individual tasks failed with OutOfMemoryError while their executor's peak memory sat right up against its ceiling, that is the second, and the Executors tab will usually show that executor marked LOST.
Common mistake: reading a FetchFailedException as a memory problem. It is a shuffle-fetch failure: a task could not read the shuffle files it needed, usually because the executor that wrote them has since died. It is a symptom of something else, often an executor OOM one stage earlier, so counting every failed task as an OOM overstates the memory story and sends you tuning the wrong knob.
Interview nuance: "explain the driver and executor model" and "your job OOMed, what do you check first" are asked at every level, and Databricks asks them of interns. The answer that scores is the fork, not the definitions: name driver OOM as a collect() problem and executor OOM as a partitioning or cores-per-executor problem, and say which metric you would look at to tell them apart.
On a real platform this differs. Here you query two small tables. In production the same rows come from the Spark UI's Executors and Tasks tabs (served live on port 4040, archived to the history server afterwards, and on EMR reachable through the persistent application UI). On AWS Glue you do not size executors directly: you buy DPUs, where one standard DPU is 4 vCPU and 16 GB billed at 0.44 USD per DPU-hour with a one-minute minimum, and Glue derives the executor layout from that. The reasoning is unchanged, only the knob has a different name.
CREATE TABLE spark_executors (
executor_id TEXT,
host TEXT,
cores INTEGER, -- task slots: one task occupies one core for its whole duration
memory_mb INTEGER, -- heap for the WHOLE process, shared by every core on it
status TEXT -- ACTIVE | LOST (LOST = the JVM died and the cluster manager noticed)
);
INSERT INTO spark_executors (executor_id, host, cores, memory_mb, status) VALUES
('driver', 'ip-10-0-4-10', 2, 4096, 'ACTIVE'),
('exec-1', 'ip-10-0-4-21', 4, 8192, 'ACTIVE'),
('exec-2', 'ip-10-0-4-22', 4, 8192, 'ACTIVE'),
('exec-3', 'ip-10-0-4-23', 2, 8192, 'ACTIVE'),
('exec-4', 'ip-10-0-4-24', 8, 8192, 'ACTIVE'),
('exec-5', 'ip-10-0-4-25', 4, 8192, 'LOST');
CREATE TABLE spark_tasks (
task_id TEXT,
stage_id INTEGER,
executor_id TEXT,
duration_ms INTEGER,
peak_memory_mb INTEGER, -- the most heap this one task held at once
status TEXT, -- SUCCESS | FAILED
error TEXT -- the exception class when it failed; NULL when it succeeded
);
INSERT INTO spark_tasks (task_id, stage_id, executor_id, duration_ms, peak_memory_mb, status, error) VALUES
('t-001', 0, 'exec-1', 4200, 1800, 'SUCCESS', NULL),
('t-002', 0, 'exec-1', 4600, 1900, 'SUCCESS', NULL),
('t-003', 0, 'exec-1', 4100, 1750, 'SUCCESS', NULL),
('t-004', 1, 'exec-1', 9100, 3100, 'SUCCESS', NULL),
('t-005', 1, 'exec-1', 8800, 3050, 'SUCCESS', NULL),
('t-006', 1, 'exec-1', 9400, 3200, 'SUCCESS', NULL),
('t-007', 0, 'exec-2', 4300, 1820, 'SUCCESS', NULL),
('t-008', 0, 'exec-2', 4500, 1860, 'SUCCESS', NULL),
('t-009', 1, 'exec-2', 9000, 3080, 'SUCCESS', NULL),
('t-010', 1, 'exec-2', 15200, 3400, 'FAILED', 'FetchFailedException'),
('t-011', 1, 'exec-2', 15800, 3450, 'FAILED', 'FetchFailedException'),
('t-012', 0, 'exec-3', 4400, 1840, 'SUCCESS', NULL),
('t-013', 0, 'exec-3', 4250, 1790, 'SUCCESS', NULL),
('t-014', 1, 'exec-3', 8900, 3020, 'SUCCESS', NULL),
('t-015', 1, 'exec-3', 9200, 3150, 'SUCCESS', NULL),
('t-016', 0, 'exec-4', 5100, 2100, 'SUCCESS', NULL),
('t-017', 0, 'exec-4', 5300, 2200, 'SUCCESS', NULL),
('t-018', 0, 'exec-4', 5000, 2050, 'SUCCESS', NULL),
('t-019', 1, 'exec-4', 12400, 7800, 'SUCCESS', NULL),
('t-020', 1, 'exec-4', 18600, 7950, 'FAILED', 'OutOfMemoryError'),
('t-021', 1, 'exec-4', 19100, 7900, 'FAILED', 'OutOfMemoryError'),
('t-022', 1, 'exec-4', 17800, 7880, 'FAILED', 'OutOfMemoryError'),
('t-023', 1, 'exec-4', 16200, 5400, 'FAILED', 'FetchFailedException'),
('t-024', 1, 'exec-5', 21000, 8140, 'FAILED', 'OutOfMemoryError'),
('t-025', 1, 'exec-5', 20400, 8100, 'FAILED', 'OutOfMemoryError'),
('t-026', 1, 'exec-5', 19800, 8050, 'FAILED', 'OutOfMemoryError');-- The Executors tab. Note the driver row: it schedules everything and runs nothing.
SELECT e.executor_id,
e.cores,
e.status,
COUNT(t.task_id) AS tasks_run,
SUM(CASE WHEN t.status = 'FAILED' THEN 1 ELSE 0 END) AS tasks_failed
FROM spark_executors e
LEFT JOIN spark_tasks t ON t.executor_id = e.executor_id
GROUP BY e.executor_id, e.cores, e.status
ORDER BY tasks_run DESC, e.executor_id;Apply
Your turn
The task this lesson builds to.
Write a query that returns every executor that ran at least one task, with its core count, its task count, and its average task duration, as (executor_id, cores, task_count, avg_duration_ms), busiest first, over spark_executors(executor_id, host, cores, memory_mb, status) and spark_tasks(task_id, stage_id, executor_id, duration_ms, peak_memory_mb, status, error).
Join the two tables on executor_id, which drops the driver row on its own because the driver runs no tasks. Round avg_duration_ms to 1 decimal, and order by task_count descending.
3 hints and 1 automated check are waiting in the workspace.
Practice
Make it stick
A second problem on the same idea, plus 3 bonus drills.
Write a query that returns every executor that failed at least one task, with its failed-task count and the share of those failures caused by an OutOfMemoryError, as (executor_id, failed_tasks, oom_pct), most failures first, over the same spark_tasks table.
Count only rows whose status is 'FAILED'. oom_pct is the percentage of that executor's failures whose error is exactly 'OutOfMemoryError', rounded to 1 decimal. Not every failure is a memory failure, so the percentage has to be computed and not assumed. Order by failed_tasks descending.
3 hints and 1 automated check are waiting in the workspace.