Apache Spark architecture: what actually happens when you call an action

• Architecture: what are the moving parts?
• Is this a master-worker architecture?
• Who does what, step by step?
  • What does the driver do?
  • What does an executor do?
  • What does the cluster manager do?
• How does one action become tasks?
• What is an RDD, and does it still matter?
• Why do stages break where they do?
• Where does a task actually run?
• Where does the driver run, and does it matter?
• Where does the memory go?
• What happens during a shuffle?
• Who turns the plan into code?
• Reading it from a running job
• When is this not the right engine?
• Production tips
• Frequently asked questions
• Conclusion
• References
• Trademarks

TL;DR

• The cluster manager is only a resource broker. Once executors register, the driver schedules every task itself and executors exchange shuffle blocks directly, which is why losing the driver kills the application and losing one executor does not.
• A job is one action, tasks are partitions, and stage boundaries fall exactly where a ShuffleDependency appears in the lineage. Everything else pipelines into one stage.
• SortShuffleManager picks between three shuffle writers in a fixed order, cheapest first. A map-side combine, which is what reduceByKey does, disqualifies the two fast ones immediately.
• Execution and storage share one memory pool, but the eviction is one-way: execution can evict cached blocks, storage can never evict execution.
• Spark 4 removed Mesos. Standalone, YARN and Kubernetes are the cluster managers now.

Most Spark tuning advice is a list of configs with no map of what they touch. You raise spark.sql.shuffle.partitions because a post said so, and when it does not help you have no way to reason about why, because the thing the knob acts on is invisible.

This is the map. It follows one count() from the line you typed to the bytes on an executor’s disk, naming the class responsible at each step so you can read the source when the behaviour surprises you. By the end you should be able to look at a slow stage in the Spark UI and say which component is struggling, rather than which config to try next.

Written against the latest Spark release, 4.2.0 as of now. Every class name, config key and default below was read from the v4.2.0 tag rather than recalled. Assumed knowledge: you have written Spark jobs and know what a DataFrame is. No internals knowledge is assumed, and every component is defined on first use.

Architecture: what are the moving parts?

The division worth internalising is that the cluster manager never schedules your work. It hands out containers, executors register back with the driver, and from that point the driver drives everything. That single fact explains a lot of otherwise puzzling behaviour: why an under-provisioned driver throttles a large cluster, why collect() on a big result kills the application rather than an executor, and why YARN’s own scheduler settings have no effect on how your tasks are distributed.

Is this a master-worker architecture?

Yes, in the usual sense: one process coordinates and many processes do the work. But Spark’s own vocabulary is worth getting exactly right, because the word “master” is overloaded and people conflate three different things with it.

Spark has also moved away from the older terminology in its own scripts. At v4.2.0 there is sbin/start-workers.sh and conf/workers.template; start-slaves.sh and slaves.template are gone, and the standalone documentation speaks of “a master and workers”.

The analogy breaks in one place worth knowing. In a classic master-worker system, the master assigns work. In Spark the cluster manager does not assign your tasks: it grants containers, and the driver schedules into them. So the coordinator role is split, resources from one component and work from another, which is why tuning YARN’s scheduler changes what you are given but never how your tasks are distributed.

Who does what, step by step?

Three components, and the clearest way to hold them is by what each is responsible for and what it explicitly is not.

What does the driver do?

The driver is where all the thinking happens. In order, for every application:

1. Creates the SparkSession, which brings up SparkContext, SparkEnv, the block manager and the RPC endpoints the executors will call back on.
2. Asks for executors. SchedulerBackend requests resources from whichever cluster manager --master named. Static allocation asks once for spark.executor.instances, which has no default in core and is documented as 2 on YARN.
3. Turns your code into a plan. Unresolved logical plan, then analysed against the catalog, then optimised by Catalyst, then a physical plan chosen by cost.
4. Cuts the plan into stages. DAGScheduler walks the RDD lineage and breaks it at every ShuffleDependency, producing a DAG of stages and a TaskSet per stage.
5. Places every task. TaskSchedulerImpl and TaskSetManager decide which executor runs which task, preferring the best available locality level.
6. Tracks every attempt. Failed tasks are retried up to spark.task.maxFailures, which defaults to 4. Slow tasks may be speculatively duplicated. A stage whose shuffle output was lost is resubmitted.
7. Keeps the block index. BlockManagerMaster on the driver knows where every cached partition and shuffle block lives, so executors can fetch from each other.
8. Receives results. Task results come back through the scheduler, bounded in total by spark.driver.maxResultSize, which defaults to 1g.
9. Serves the UI on port 4040 while the application runs, and writes the event log the history server replays afterwards.

The driver defaults to spark.driver.memory of 1g, which is fine for a coordinator and not fine for anything that pulls data back. It is also the one process with no redundancy: if the driver dies the application dies, because the plan, the scheduler state and the block index all die with it.

What does an executor do?

Much less, deliberately. The source describes it in one line as a “Spark executor, backed by a threadpool to run tasks”. The lifecycle is:

1. Registers with the driver. CoarseGrainedExecutorBackend sends RegisterExecutor and waits for RegisteredExecutor before it will accept work. Until that round trip completes, a granted container is doing nothing.
2. Accepts LaunchTask messages and runs each one in a thread from the pool. The number of concurrent tasks is spark.executor.cores, which defaults to 1 on YARN and to every core on the machine in standalone mode.
3. Reads its input, runs the generated code over it, and for a shuffle map stage writes shuffle output to local disk.
4. Reports back with StatusUpdate per task, and sends a heartbeat every spark.executor.heartbeatInterval, which defaults to 10s. Miss enough of those and the driver declares the executor lost.
5. Stores and serves blocks through its BlockManager, which is what makes a cached DataFrame reusable and what lets the next stage fetch shuffle data from its peers.

An executor never makes a scheduling decision, never sees the query plan, and never talks to another executor about what to run. It receives tasks, runs them, and serves bytes. That is the whole contract, and it is why executors are disposable: lose one and its tasks are retried elsewhere, with cached partitions recomputed from lineage.

What does the cluster manager do?

Exactly one thing: decide who gets containers. It is a resource manager rather than a work scheduler, and how it does the job is the only part that changes between deployments.

Kubernetes is the clearest illustration of the split, because there is no Spark cluster-manager daemon at all: the driver talks to the API server directly and asks for pods. The component granting resources and the component scheduling work are simply different things, however they happen to be packaged.

Two defaults follow from this. spark.dynamicAllocation.enabled is false, so absent a decision your application holds its executors from first request to exit, idle or not. And Mesos support was removed in the 4.x line, so the three rows above are the whole list.

And the end-to-end flow, from the line you type to the result you get back:

flowchart TB
  S1["<b>1</b>  you call an action<br/><i>count, collect, write</i>"] --> S2["<b>2</b>  driver builds the logical plan<br/><i>unresolved, then analysed</i>"]
  S2 --> S3["<b>3</b>  Catalyst optimises and<br/>selects a physical plan"]
  S3 --> S4["<b>4</b>  DAGScheduler cuts stages<br/><i>at every ShuffleDependency</i>"]
  S4 --> S5["<b>5</b>  driver asks the cluster manager<br/>for executors, if it has none"]
  S5 --> S6["<b>6</b>  TaskSchedulerImpl places tasks,<br/><i>best locality first</i>"]
  S6 --> S7["<b>7</b>  executors run tasks and write<br/>shuffle output for the next stage"]
  S7 --> S8{"more stages?"}
  S8 -->|"yes"| S6
  S8 -->|"no"| S9["<b>8</b>  results return to the driver,<br/>or are written by the executors"]

Step 8 has a fork worth noticing. collect() pulls every row back to the driver, which is why it is the usual cause of driver memory failures. write does not: each executor writes its own partition directly to storage and returns only a status. Same job shape, completely different driver cost.

Steps 6 and 7 repeat per stage. That loop is where adaptive query execution intervenes, because between two passes it has real statistics from the stage that just finished.

How does one action become tasks?

Nothing runs until an action. filter, select and join add to a plan; count, collect, show and write execute it. When an action fires, DAGScheduler walks the plan backwards and cuts it into stages wherever the data has to move between partitions.

flowchart LR
  subgraph ST1["stage 0: narrow, no shuffle"]
    direction LR
    R["read parquet<br/>8 partitions"] --> F["filter"]
    F --> W["withColumn"]
  end

  subgraph ST2["stage 1: after the shuffle"]
    direction LR
    AG["aggregate"] --> WR["write, the action"]
  end

  W -->|"shuffle: groupBy city_id"| AG

Every stage but the last is a ShuffleMapStage: its job is to produce shuffle output for the next stage. The last is a ResultStage, which computes the answer the action asked for. That is why the Spark UI shows a final stage with a different shape from the rest.

A concrete count, on a table that reads as 8 partitions:

from pyspark.sql.functions import col, when

trips = spark.read.parquet("s3a://lakehouse-prod/warehouse/trips")   # 8 partitions

by_city = (trips
    .filter(col("fare_amount") > 50)                                 # narrow
    .withColumn("band", when(col("fare_amount") > 100, "high")
                        .otherwise("mid"))                           # narrow
    .groupBy("city_id").count())                                     # WIDE

by_city.write.parquet("s3a://lakehouse-prod/warehouse/trips_by_city")

One action, so one job. One wide transformation, so two stages. Eight input partitions, so eight tasks in stage 0, and spark.sql.shuffle.partitions tasks in stage 1 unless adaptive execution coalesces them.

What is an RDD, and does it still matter?

Everything above sits on one abstraction. A Resilient Distributed Dataset is a partitioned, immutable collection that knows how to rebuild itself. RDD.scala states the five properties that define one:

The source is explicit that this is not incidental: “All of the scheduling and execution in Spark is done based on these methods.” Every component in this post reads one of those five.

Resilient is the third property doing work. An RDD records what it was computed from, which is the lineage mentioned earlier. Lose a partition and Spark replays the function against the parent rather than checkpointing state, which is why executor loss is survivable and driver loss is not: lineage lives on the driver.

You almost certainly do not write RDD code any more, and that is fine. A DataFrame is a higher-level API that Catalyst optimises and then compiles down to exactly these objects, which is why df.rdd.getNumPartitions() works and why the Spark UI still talks about RDDs. The abstraction did not go away; it stopped being the thing you type.

Why do stages break where they do?

Stage boundaries are not a heuristic. They fall out of the dependency type between one RDD and its parent, and Dependency.scala defines exactly two families.

NarrowDependency, in the source’s own words, is where “each partition of the child RDD depends on a small number of partitions of the parent RDD”, and it notes that narrow dependencies “allow for pipelined execution”. Two concrete subclasses:

ShuffleDependency is the other family: a dependency on the output of a shuffle stage, carrying the Partitioner that decides which reducer each record belongs to.

flowchart TB
  subgraph NW["narrow: pipelined into one stage"]
    direction LR
    N1["p1"] --> N1b["p1'"]
    N2["p2"] --> N2b["p2'"]
    N3["p3"] --> N3b["p3'"]
  end

  subgraph WD["wide: a ShuffleDependency, so a new stage"]
    direction LR
    W1["p1"] --> X1["p1'"]
    W1 --> X2["p2'"]
    W2["p2"] --> X1
    W2 --> X2
    W3["p3"] --> X1
    W3 --> X2
  end

  NW ~~~ WD

That is the whole rule. DAGScheduler walks the lineage backwards, and every ShuffleDependency it meets ends a stage. Counting the wide transformations in your code therefore predicts your stage count before you run anything, and it is why a broadcast join is worth forcing: it converts a ShuffleDependency into a narrow one, removing a stage boundary entirely.

Where does a task actually run?

TaskSchedulerImpl does not place tasks arbitrarily. It offers each task the best locality it can get, and waits a little before settling for worse. TaskLocality defines five levels, in this order:

spark.locality.wait (default 3s) is how long the scheduler holds out for a better level before giving up and taking the next one. That default is the reason a job on a busy cluster sometimes appears to stall between stages: the scheduler is waiting for a local slot rather than doing nothing.

Each executor runs spark.executor.cores tasks at once, and each task takes spark.task.cpus (default 1) of them. The two together, not the executor count alone, decide your real parallelism.

Where does the driver run, and does it matter?

The cluster manager decides where executors go. --deploy-mode decides where the driver goes, and that choice has consequences the other articles on this topic tend to skip.

SparkSubmit rejects combinations that cannot work, and the error text is worth recognising:

Cluster deploy mode is not compatible with master "local"
Cluster deploy mode is not applicable to Spark shells.
Cluster deploy mode is currently not supported for python applications on standalone clusters.

The first is a category error: local has no cluster to put a driver in. The second is the interesting one, and it explains why every interactive session is client mode. A shell needs a REPL attached to the driver, so the driver has to be where you are. That in turn is why a collect() sized for a production cluster can exhaust the heap on your laptop, and why closing the laptop ends the job.

One piece of vocabulary these diagrams often blur: a worker node is a machine in the cluster, while an executor is a JVM running on one. A node can host several executors, and spark.executor.cores is slots per executor, not per node. Sizing decisions are about executors; capacity planning is about nodes.

Where does the memory go?

The formula is in UnifiedMemoryManager: subtract a fixed 300 MB, then take spark.memory.fraction of the rest. For --executor-memory 16g on the defaults, that leaves 9,650 MB shared between shuffles, joins, sorts and cache combined, and asks the cluster for 18,022 MB once overhead is added.

The behaviour that matters more than the numbers is the asymmetry. Execution and storage draw from one pool and borrow from each other, but execution can evict cached blocks down to the storage floor, and storage can never evict execution. Caching a large DataFrame therefore cannot fail a shuffle, but it can make one spill, which is the usual reason a job gets slower after someone adds a cache().

What happens during a shuffle?

A shuffle is the expensive thing every other decision is arranged around. The map side writes partitioned files, the reduce side fetches them.

SortShuffleManager.registerShuffle picks between three writers in a fixed order, and the conditions are worth knowing because they are checkable:

The condition that disqualifies both fast paths is mapSideCombine. That is what reduceByKey and aggregateByKey set, which is the trade hiding behind the usual advice to prefer them over groupByKey: they shuffle far less data but give up the cheaper writers.

Knobs on the fetch side:

That last default is the one to change deliberately. Without the external shuffle service, dynamic allocation cannot release an executor that still holds shuffle blocks another stage needs.

Who turns the plan into code?

Everything above is the runtime. The plan it runs comes from Catalyst, which takes a query through four phases, each a set of rules over a tree.

flowchart LR
  SQL["SQL or DataFrame"] --> AN["<b>Analyzer</b><br/>resolve names<br/>against the catalog"]
  AN --> OPT["<b>Optimizer</b><br/>pushdown, pruning,<br/>constant folding"]
  OPT --> PLAN["<b>SparkPlanner</b><br/>strategies produce<br/>physical candidates"]
  PLAN --> CG["<b>WholeStageCodegenExec</b><br/>fuse operators into<br/>one Java method"]
  CG --> RDD["RDD[InternalRow]"]
  RDD -.->|"runtime statistics"| AQE["<b>AdaptiveSparkPlanExec</b>"]
  AQE -.-> PLAN

The dashed loop is adaptive query execution, on by default since 3.2. After a stage completes, real statistics go back into planning, so the next stage’s join strategy and partition count come from measured sizes rather than estimates. It is why spark.sql.shuffle.partitions matters far less than it used to: AQE coalesces toward spark.sql.adaptive.advisoryPartitionSizeInBytes, which falls back to 64MB.

Catalyst can only optimise what it can see. A SQL expression is a tree it can rewrite; a Python UDF is an opaque function it must call as-is, which blocks pushdown across that point. That is the real cost of a UDF, and it is usually larger than the cost of the function itself.

Reading it from a running job

Every claim above is observable. This prints the structure the sections describe, without leaving the shell:

# How many partitions am I actually working with?
print("input partitions:", trips.rdd.getNumPartitions())

# Which plan did Catalyst settle on? "formatted" is the readable one.
by_city.explain("formatted")

# Did a join get broadcast, and was the filter pushed into the scan?
# Look for BroadcastHashJoin and PushedFilters in the output above.

# What is this executor's memory split, from the running JVM?
conf = spark.sparkContext.getConf()
for k in ("spark.executor.memory", "spark.executor.cores",
          "spark.memory.fraction", "spark.memory.storageFraction",
          "spark.sql.shuffle.partitions", "spark.sql.adaptive.enabled"):
    print(f"{k:42s} {conf.get(k, '(default)')}")

Which brings us back to the promise at the top. Read a slow stage this way:

In the Spark UI, three pages answer most questions. The stages page gives task duration percentiles and spill columns, which is where skew shows up. The SQL tab gives the physical plan annotated with row counts, which answers “did my filter push down”. The storage tab shows what is cached and what fraction of it actually fits in memory.

When is this not the right engine?

Spark’s architecture assumes work worth distributing. Where that does not hold, the machinery is overhead you are paying for nothing.

Data that fits in memory on one machine. A few gigabytes is a pandas, DuckDB or Polars job. Spark’s scheduling, serialisation and shuffle machinery costs more than the computation saves.

Low-latency single-row lookups. Spark plans a job per action, with tens of milliseconds of overhead before any work starts. A key-value store or an indexed database answers those; Spark scans.

Very many tiny jobs. Each action pays planning and scheduling. Thousands of small jobs spend their time in the driver, which is why per-request work belongs in a service rather than a Spark application.

Spark earns its cost when the data does not fit on one machine, the work is expressible as transformations over partitions, and the job runs long enough for planning overhead to be a rounding error.

Production tips

• Size the container, not the heap. The cluster manager must grant spark.executor.memory plus overhead, roughly 10 percent more. Sizing to the nearest gigabyte fails to schedule.
• Set spark.executor.cores deliberately. Four to five is the usual range; all the tasks on an executor share one heap, so more cores means less memory each.
• Turn on the external shuffle service before dynamic allocation. Without it, executors holding shuffle blocks cannot be released.
• Leave adaptive execution on. It coalesces partitions and re-picks join strategies from real statistics, and it is on by default.
• Prefer reduceByKey to groupByKey, accepting that a map-side combine rules out the two cheaper shuffle writers. Shuffling less data is worth more.
• Watch task duration percentiles, not averages. Skew is invisible in a mean and obvious in a max.
• Keep the driver off your laptop for production runs. Client mode puts the driver where you are sitting, so a collect() that works on the cluster can exhaust your local heap.

Frequently asked questions

Why does my job have more jobs than actions? Some operations run a hidden job first. show() can run one job to peek and another to finish, and anything that samples or infers, such as reading JSON without a schema or a sortBy computing range boundaries, runs its own job to collect what it needs.

Losing an executor: what survives? Its cached blocks are gone and will be recomputed from lineage, the record Spark keeps of the transformations that produced each partition. Because that record is held on the driver, any lost partition can be rebuilt by replaying the steps that made it. Its shuffle output is gone too, unless the external shuffle service is holding it, in which case DAGScheduler can skip resubmitting the map stage.

Why is the driver a single point of failure but an executor is not? The driver holds the plan, the scheduler state and the block index. An executor holds only data that can be recomputed. There is no driver failover in the application itself; cluster-mode restart policies are what cover it.

Where does off-heap memory fit? spark.memory.offHeap.enabled is false by default and spark.memory.offHeap.size is 0. When enabled, Tungsten, Spark’s binary memory and code-generation layer, allocates outside the JVM heap. That takes pressure off garbage collection, at the cost of a second budget to manage.

Is Mesos still supported? No. It was removed in the Spark 4 line; the resource manager module is gone entirely. Standalone, YARN and Kubernetes are the options.

What is Spark Connect and does it change this picture? It splits the client from the driver: your application sends an unresolved plan over gRPC and a remote driver runs everything described above. The runtime is unchanged; what moves is where your code lives relative to the driver JVM.

Conclusion

Back to the config you raised without knowing what it touched. Almost every Spark tuning decision lands on one of three things: how many partitions there are, how large each one is, and whether the data has to move between them. The components in this post are just the machinery that answers those three questions, which is why a map of them is more durable than a list of settings.

The division that explains the most is between the cluster manager and the driver. The manager hands out containers and then leaves; the driver schedules every task, tracks every block, and holds state that nothing else can rebuild. Once that is clear, the behaviours that look arbitrary stop being arbitrary: the driver as a bottleneck, the driver as a single point of failure, and the cluster manager’s own scheduler settings having no effect on your task placement.

Read the plan before the configs. explain("formatted") and the SQL tab tell you what Catalyst decided, the stages page tells you how that decision played out across partitions, and only then does a config change have a target. A knob turned without that is a guess, however good the post it came from.

References

• Spark configuration reference for every property, its default and the version it appeared in
• DAGScheduler.scala at v4.2.0, the stage-oriented scheduling layer described above
• RDD.scala at v4.2.0 for the five properties every scheduling decision reads
• SortShuffleManager.scala at v4.2.0 for the writer selection and its conditions
• Apache Spark: the complete cheat sheet for the config surface this post explains

Trademarks

Apache Spark, Apache Hadoop, Apache Mesos, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries. Kubernetes is a registered trademark of The Linux Foundation.