The Spark Execution Model
A driver that plans and schedules, executors that hold data and run tasks, and a cluster manager that hands out machines. Almost every confusing Spark failure is explained by knowing which of the three it happened on.
Who needs this, what one row is, and why the obvious build breaks
Every lesson starts from the consumer, because designing from the source outward is this domain's characteristic mistake.
When a Spark job runs, what is running where — and which machine is the one that just ran out of memory?
The engineer on call for a job that failed at 03:40, holding a stack trace and a cluster UI. Whether they fix it in ten minutes or three hours depends entirely on whether they can tell a driver failure from an executor failure from a task failure.
Four nested units, and mixing them up is the source of most confusion. An application is one driver and its executors. A job is one action. A stage is a set of tasks with no shuffle between them. A task is one stage's work on one partition — the smallest unit that is scheduled, retried and measured.
Treat the cluster as a black box that runs your code. Write the transformation, submit it, and read the error message when it fails. This works for a long time, because most jobs succeed and most failures are genuine bugs in the transformation.
The job dies with an out-of-memory error and the instinct is to increase executor memory. The failure was on the driver, because the code called an action that pulls every row back to it, and executor memory is irrelevant to it (Lazy Evaluation).
- The job dies with an out-of-memory error and the instinct is to increase executor memory. The failure was on the driver, because the code called an action that pulls every row back to it, and executor memory is irrelevant to it (Lazy Evaluation).
- A closure references an object from the surrounding scope — a dictionary, a client, a configuration blob — and that object is serialized and shipped with every task. The job is slow for a reason that appears nowhere in the query.
- Code that works in a local shell fails on the cluster because it reads a file from the local filesystem. On one machine driver and executor are the same process; on a cluster they are not, and the file exists on neither of the other twenty machines.
- The application holds a large cluster for forty minutes while the driver does something single-threaded — planning against a catalog with thousands of partitions, or looping over results — and every executor sits idle and billed.
What is actually happening
- The driver runs your program, builds the logical plan, optimises it, cuts it into stages at shuffle boundaries, and schedules tasks onto executors. It holds the accumulated metadata for the application and it is a single point of failure: when it dies, the application dies.
- The executors are long-lived processes on worker machines, each with a number of cores and a memory budget. Each core runs one task at a time, so an executor with four cores is four task slots. Executors also hold cached data and serve shuffle blocks to other executors (Worker Pools Beyond Threads).
- A cluster manager — YARN, Kubernetes, or the engine's own standalone mode — decides which machines the application gets. It is a resource allocator and knows nothing about your query (Scheduling: How a Pod Chooses a Node).
- Calling an action triggers a job. The driver walks the plan backwards from the action, cuts it at each shuffle, and submits the earliest stage first. Tasks in a stage are independent; stages are not (Stages and Tasks).
- Everything the task needs — the closure, the broadcast variables, the referenced objects — is serialized on the driver and sent to executors. This is why a job can be slow because of what a closure happened to capture, and why non-serializable objects fail at submission rather than at compile time.
Three roles, and which one just failed
Every confusing failure in a distributed job starts with an unasked question: which machine did this happen on? The driver, an executor, and the cluster manager fail in different ways, produce different errors, and are fixed by different changes — and the error message rarely says which one it was.
The driver is your program. It is where the plan is built, where scheduling decisions are made, where results come back if you ask for them, and where broadcast data is assembled before being sent out. It is one machine, and it is not replaceable mid-run.
Executors are the workers. They are told which partition to process and with which closure, they run it, and they hold on to shuffle output and cached blocks afterwards so other tasks can fetch them. Losing one costs work; losing many costs a stage.
| Trigger | Symptom | Cause | Response |
|---|---|---|---|
| An action returns every row to the program. | Driver out of memory, often after every task has succeeded. | The result set was materialised on one machine that was sized for planning, not for data. | Write from the executors to storage and read the output separately. Reserve result-returning actions for genuinely small outputs. |
| One partition is far larger than the others. | A single executor out of memory, or one task running for many multiples of the median. | A hot key concentrated rows into one task, whose memory budget is per task rather than per job. | Fix the distribution rather than the memory: broadcast the small side, split the hot key, or pre-aggregate (Salting a Skewed Key). |
| The node running an executor is pre-empted. | A stage retries; sometimes an *earlier* stage recomputes. | The lost executor was also serving shuffle output, so the data the next stage needed no longer exists. | Use an external shuffle service where available, and prefer fewer, larger stages over long chains of small ones (The Shuffle). |
| A closure captures a large or non-serializable object. | Task serialization error at submission, or a job that is inexplicably slow to start each stage. | Everything a task references travels with it, once per task. | Broadcast it once per executor, or construct it inside the task rather than capturing it from the enclosing scope. |
From one action to thousands of tasks
Nothing runs when you write a transformation. The driver accumulates a plan, and an action — writing, counting, returning rows — is what turns that plan into scheduled work (Lazy Evaluation).
The unit hierarchy below is worth memorising, because every number in the cluster UI is reported at one of these levels and reading a stage-level number as if it were job-level is how people conclude the wrong thing about a job. The parallelism you actually get is decided at the bottom of it: tasks per stage.
The hierarchy also explains retry behaviour. A task retries cheaply. A stage retry re-runs every task in it. An application failure re-runs everything not already committed — which is the argument for committing in bounded ranges rather than at the end of a six-hour job.
| Unit | Created by | Runs on | What its failure costs |
|---|---|---|---|
| Application | Submitting the program | Driver + allocated executors | Everything not already committed to storage. |
| Job | One action | The whole cluster | The action re-runs, including every stage it depends on that was not cached. |
| Stage | A shuffle boundary in the plan | All executors, in parallel tasks | Every task in the stage re-runs; if the shuffle inputs were lost with an executor, the previous stage too. |
| Task | One partition of one stage | One core slot on one executor | A retry, usually invisible — and the reason non-deterministic tasks are dangerous. |
Application ── one driver + its executors, alive for the session
Job ── one action (write, count, collect)
Stage ── work with no shuffle inside it
Task ── one stage's work on one partition <-- the scheduled unit
|
└─ runs in one core slot on one executor
retried on failure, on another executor if this one is goneAdaptive execution — the engine re-planning a stage at runtime using the statistics its shuffle just produced, for example coalescing small partitions or splitting a skewed one — has been part of Spark for several major versions and is on by default in recent ones. Whether it is enabled, and what it will do for your particular skew, is a version and configuration question: check the documentation for the version you actually run.
Locality, and why it usually does not save you
The original design assumed compute ran on machines that held the data on local disk, so scheduling a task near its data avoided the network entirely. On a modern platform where data lives in object storage and compute is elastic, that locality is mostly gone: every task reads across the network by default (Separating Storage from Compute).
What replaced it is a different economy. Reading from object storage is bandwidth-limited and request-limited rather than latency-limited, so the wins come from reading fewer bytes and fewer objects rather than from reading them from a closer disk. This is exactly why layout decisions dominate: pruning a partition removes a read entirely, and no amount of scheduling cleverness competes with that (Partition Pruning).
Locality does still matter in one place, and it matters a great deal: shuffle blocks. Those are written to local disk on the executor that produced them and fetched over the network by the next stage. When an executor dies, its blocks die with it — which is why executor churn is more expensive than it looks and why an external shuffle service exists at all.
Double the executor count and the memory per executor, resubmit, and see whether the wall clock improves.
Find the stage taking the time. Compare its max task duration with its median. Look at its shuffle read bytes and its spill. Then choose: fewer bytes read (layout, projection), fewer bytes moved (join strategy, pre-aggregation), or more even tasks (skew handling). Change the cluster only when the tasks are already even and there are more of them than slots.
Extra capacity only helps the one condition where tasks are uniform and queued behind slot availability. When the stage is bound by one large task, by object-storage requests, or by driver-side work, adding executors adds cost with no effect — and the cluster UI says which of those it is before you spend anything.
How to build it
Most important first.
- Keep the driver out of the data path. Actions that materialise results on the driver belong to debugging and to genuinely small outputs; everything else should be written from the executors to storage.
- Size executors so that a task has enough memory for its partition with room for the shuffle and the aggregation, and prefer a moderate number of cores per executor over one enormous executor — the memory is shared by every task in the process (Sizing a Thread Pool).
- Broadcast anything a task needs repeatedly rather than letting a closure capture it, so it is shipped once per executor instead of once per task (Broadcast Joins).
- Read the plan and the stage graph before changing configuration. The stage count, the shuffle sizes and the task duration distribution answer most questions that people instead answer by doubling the cluster (Reading EXPLAIN ANALYZE).
- Give the driver enough memory for planning against a catalog with many partitions and files. Driver pressure is often metadata pressure rather than data pressure (File Size and the Small-Files Problem).
What this actually promises
Naming the guarantee you do not have is worth more than naming the one you do — everything downstream inherits the weakest promise in the chain.
- A failed task is retried, on a different executor if one has been lost. That guarantee is what makes long jobs survivable, and it is conditional on the task being a deterministic function of its input.
- The driver is not fault-tolerant in the way tasks are. Losing it loses the application and everything not yet committed to storage.
- Nothing promises that a retried task has no external side effects. If the task wrote to an external system directly, the write may have happened before the retry (Idempotent Data Pipelines).
- Task scheduling promises locality preference, not locality. When no slot is available near the data, the scheduler will run the task elsewhere and read across the network instead (Load Balancing).
Can I trust it?
A green pipeline is evidence that code ran. These four fields are the evidence that the data is right.
- Check the output row count and a summed measure against the source for the period, from a separate query rather than from the job's own accumulators. A job's internal counters describe what the job believes it did (Reconciliation).
- Alert on task failure counts even when the job succeeded. A stage that needed forty retries to finish is a job that is about to stop finishing, and it is invisible in a green run (Pipeline Observability).
- Neither check sees a task that succeeded on retry after a partial external write. That failure is only visible in the external system.
- Application startup — acquiring executors, initialising them, planning — is a fixed cost paid before any data is touched, and it sets the practical floor on how short a batch job can usefully be.
- A long-lived application that serves many small jobs amortises that cost, which is why interactive and scheduled workloads often want completely different cluster arrangements (Cost vs Freshness).
- Freshness of the output is unaffected by any of this. What the model decides is how much of the interval is spent on overhead rather than work.
- The driver/executor split and the task-per-partition model have been stable for the life of the engine. Configuration names, memory accounting details and the default join strategies have not, and advice that hangs on a specific setting ages badly.
- A schema change upstream is discovered by the driver at planning time when the plan is resolved against the catalog, which is the one place a distributed job fails early and loudly (Schema Evolution).
- Engine upgrades change plans. A job whose runtime depended on a particular join strategy can regress on upgrade without any code change, which is an argument for asserting on data and duration rather than on the plan (Pipeline SLOs).
- Task and stage retries are automatic. Their cost is that a stage which lost its inputs must recompute them, so a late failure can rewind further than it appears to.
- An application that dies leaves whatever its sink committed. With a table format that swaps metadata atomically, the answer is "nothing partial"; with a directory of files, the answer is "some of it" (Atomic Publish).
- Restarting the application re-runs the whole job unless the pipeline is chunked into independently committed ranges. Long jobs should be chunked for exactly this reason (Incremental Processing).
What can go wrong
- Driver out of memory from collecting results, from broadcasting something large, or from planning against enormous catalog metadata.
- Executor lost to the cluster manager — pre-emption, node failure, container limits — taking its cached blocks and shuffle output with it (OOM Kills and CPU Throttling).
- Serialization failure at submission because a closure captured something that cannot be shipped, which appears as a confusing error far from the offending line.
- A stage that repeatedly retries because a fetch fails, and eventually fails the job after exhausting its attempts (The Shuffle).
- The mitigation failing: increasing executor memory when the pressure was on the driver, which changes cost and nothing else.
- "Out of memory means I need bigger executors." It means some JVM ran out. Which one, and doing what, changes the fix completely.
- "The cluster has 200 cores so my job runs 200 ways parallel." It runs as parallel as it has partitions. A job with twelve partitions uses twelve slots and leaves the rest idle (Partitions: the Unit of Parallelism).
- "The driver is just a launcher." It plans, schedules, tracks every task, receives every result you ask for, and holds broadcast data. It is the most stateful machine in the application.
- "More executors always finish faster." Beyond the number of tasks in the widest stage, extra executors add scheduling and shuffle-fan-out cost and nothing else (Why Eight Cores Give You Four and a Half).
Operating it
- Task duration distribution per stage, plus shuffle read and write bytes. These two together explain most job pathology (Tail Latency: Why p50 Being Fine Does Not Help).
- Executor lifetime and loss events over the run. Frequent executor churn is a cluster or capacity problem masquerading as a job problem.
- Driver memory and garbage-collection time, which is the only signal that separates "the query is heavy" from "the program is doing something on one machine" (Garbage Collection: Pause, Throughput, Footprint — Pick Two).
- At 10x data, nothing about the model changes: more partitions, more tasks, the same three roles.
- At 100x, the driver becomes a bottleneck in ways it never was before — plan size, partition metadata, scheduling throughput — and jobs are often split rather than grown (Partition Cardinality).
- At high executor counts, the shuffle becomes a many-to-many network problem and its metadata alone becomes significant (The Shuffle).
- Cost accrues per executor per second regardless of whether a task is running. Idle slots during a straggler or during driver-side work are paid for at full rate (Compute Waste).
- Over-provisioned executor memory is charged whether or not it is used, and is a common response to a failure that was not a memory failure.
- Startup dominates short jobs. A pipeline of thirty tiny applications pays thirty startups; one application doing thirty steps pays one.
- The driver-centric model makes planning global and powerful, and makes one machine a single point of failure for a thousand-core job.
- Long-lived executors give fast task start and cached data, and hold resources you are paying for whether or not the next job needs them.
- Automatic retry hides transient failure, which is exactly what you want operationally and exactly what makes non-deterministic transformations dangerous (Determinism: Same Input, Same Output?).
Where this applies
Almost nothing here is universal. These labels say what each claim is specific to, and where a different engine, format, warehouse or scale would differ.
- ENGINE-SPECIFICDriver, executors and a cluster manager is Spark's arrangement. Flink has a JobManager and TaskManagers with a long-lived deployed graph instead of per-action jobs; Trino has a coordinator and workers with no lineage-based recovery, so a lost worker fails the query rather than triggering recomputation.
- GENERALThe underlying split — one process that plans and coordinates, many that hold data and execute — is shared by every distributed query system, and so is the consequence that the coordinator is the fragile part.
- SIMPLIFIEDAdaptive runtime re-planning, dynamic executor allocation and external shuffle services all break the clean "plan once, then execute" story. They change when decisions are made, not which decisions exist.
Where the depth lives
This domain teaches how data moves and how you know it arrived intact. It hands the rest off by name.
- — Distributed Systems owns what a cluster manager is doing when it allocates and pre-empts resources, and why a coordinator that is not replicated is a single point of failure by design rather than by oversight.