Almost every confusing Spark failure becomes obvious once you know which process was holding the thing that ran out. This section is about that: what the driver is, what an executor is, what lives in each, and which failures belong to which.

### Two kinds of process, and only two

A Spark application is one driver and some number of executors.

The driver runs your code. It builds the logical plan, runs the optimiser, cuts the plan into stages, decides which tasks exist, hands them to executors, collects their results and decides what happens next. It is a single process, and nothing about it is distributed.

An executor is a worker process with a heap and a fixed number of task slots. It receives serialised tasks, runs them against the partitions it is told to read, holds cached blocks, serves shuffle data to other executors, and reports back. Executors do not talk to your code. They run closures your code produced.

That is the whole topology. There is no master doing computation, no coordinator holding data, and no hidden tier. Every piece of behaviour in this lesson follows from where a thing lives in that two part split.

### What the driver holds, which is more than people expect

The driver holds the plan, the task metadata for every task in flight, the broadcast variables, the accumulator values and, crucially, anything you asked to bring back.

That last one is the source of the single most common driver failure. A `collect` pulls every row of a distributed dataset into the driver's heap. A `toPandas` does the same. A `take` of a large number does the same with extra steps. The dataset was fine spread across forty executors and is not fine in one process with eight gigabytes.

The second most common is task metadata. A job with two hundred thousand tasks has two hundred thousand task descriptions, each with its locality preferences and its accumulator state, all in the driver. This is why a job with absurd partition counts can kill the driver without any data ever being collected.

The third is broadcast. A broadcast variable is built on the driver before it is sent, so a two gigabyte broadcast needs two gigabytes of driver heap plus the serialised copy, and the driver dies before any executor sees a byte.

<PositionedImage src="/uploads/courses/bd_spark_architecture_and_memory_model/01_driver_executor_topology.gif" videoUrl="https://codemia-static-assets.nyc3.cdn.digitaloceanspaces.com/uploads/courses/bd_spark_architecture_and_memory_model/01_driver_executor_topology.mp4" align="center" width="900" alt="A topology diagram showing one driver process connected to several executor processes, with the data flowing between executors rather than through the driver" alt2="" caption="Data moves between executors. The driver holds the plan and the task metadata, and the only data that reaches it is data you asked for." />

### What an executor holds

An executor's heap holds four things that compete with each other: the rows a task is currently working on, the data structures a shuffle or an aggregation needs, the cached blocks, and whatever the JVM itself needs.

The first two are what Spark calls execution memory. The third is storage memory. The fourth is overhead and is the one people forget.

Executors also hold shuffle output on local disk, which is not heap but is a resource that runs out, and the shuffle service that serves it. An executor that is removed takes its shuffle files with it unless an external shuffle service is running, which is why dynamic allocation and shuffles interact badly without one.

### Which failure belongs to which process

Four failures and where each lives, because the error messages do not say.

A driver out of memory means you collected something, broadcast something, or created too many tasks. No amount of executor memory changes it.

An executor out of memory means one task's working set did not fit, which is almost always a partition that is too large, a skewed key, or a user defined function holding something per row. Adding executors does not help, because the failing task is one task.

A container killed by the resource manager means the process exceeded its total allocation, heap plus overhead. This is not a Java heap error, it is the operating system, and the fix is in the overhead setting rather than the heap setting.

A stage failing with a fetch failure means an executor that held shuffle data is gone. The task that failed is innocent and the executor that died is the subject.

<Callout type="insight">
Before reading any Spark stack trace, find out which process produced it. The driver log and the executor log are different files, they fail for different reasons, and the same words appear in both. A large share of the time spent on Spark memory problems is spent tuning the setting that belongs to the other process.
</Callout>

### Cluster managers, briefly

The driver and executors have to be started by something, and that something is a cluster manager: YARN, Kubernetes, Mesos historically, or Spark's own standalone mode.

What varies between them is how resources are requested, how containers are sized and what happens when one dies. What does not vary is anything in this lesson. The memory model inside an executor is identical on all of them, and so is every failure above.

The one place the manager matters is the overhead setting, because the manager is what kills a container for exceeding its total. On YARN and Kubernetes the container size is heap plus overhead, and a process that uses more native memory than the overhead allows is killed without a Java error of any kind.

### Deploy modes

One more distinction worth having, because it changes where the driver's failures appear.

In client mode the driver runs where you submitted from, which is a laptop, a notebook server or an edge node. Its heap is whatever that machine gives it and its logs are on your screen. This is convenient for development and a problem in production, because the job dies when the submitting process does.

In cluster mode the driver runs inside the cluster as another container, with a size you requested. Its logs are wherever the cluster manager puts them, which is the thing people cannot find.

The failure above about task metadata behaves very differently between the two. A client mode driver on a laptop has whatever heap the laptop has, which is usually generous. A cluster mode driver has exactly what you asked for, which is usually the default, which is usually small.

### The number of executors is not fixed

One assumption worth removing early: the executor count can change while the application runs.

With dynamic allocation enabled, the driver asks for more executors when there is a backlog of pending tasks and releases them when they sit idle. This is why a job's resource usage over time is a shape rather than a rectangle, and why the cost of a job is not simply its executor count multiplied by its duration.

The interaction with shuffles is the part that bites. An idle executor released after a shuffle stage still holds that stage's output on its local disk. When a later task tries to fetch it, there is nothing there, and the stage that produced it has to run again. The external shuffle service exists precisely to break that coupling: it runs on the node rather than inside the executor, so the files outlive the process that wrote them.

Without that service, dynamic allocation and a large shuffle are a bad pairing, and the symptom is a job that mysteriously recomputes stages it already finished.

### Why the two process split explains the API

The split also explains a rule that otherwise looks arbitrary: anything you pass into a transformation has to be serialisable.

Your code runs on the driver. The function inside a `map` or a `filter` runs on an executor. For that to happen, the function and everything it closes over must be serialised, shipped and reconstructed in another process on another machine.

So a closure that captures a database connection fails, because a socket cannot be reconstructed elsewhere. A closure that captures a large lookup table succeeds but ships a copy to every single task, which is the problem broadcast variables solve. A closure that captures `self` in a class method drags the whole object along with it, which is the classic accidental serialisation of something enormous.

The error messages here are famously unhelpful, but the rule behind them is not complicated. If a value is created on the driver and used inside a task, it crosses a process boundary, and everything that cannot cross one is a failure.

### What this lesson is going to do

The rest of this lesson is inside one executor. The next section divides the heap into its regions and explains the boundary that moves between two of them. The one after that covers why a larger heap is not always better. Then slots and what runs in one, and finally how to choose executor sizes for a workload rather than from a template.

None of it is about tuning in the sense of trying values. All of it is about knowing what a setting controls, so that when a job fails you can name the region that filled and change the one thing that governs it.