What is the difference between the driver and an executor?

The driver runs your main program, builds the query plan, and schedules tasks. Executors are JVM processes on worker nodes that actually run those tasks and hold cached data. There is exactly one driver per application and usually many executors.

What does a cluster manager do that Spark cannot do itself?

It owns the machines and decides which applications get which resources. Spark asks for executors with a certain CPU and memory shape; YARN, Kubernetes, or Spark's own standalone manager decides where they run and enforces the limits. Spark schedules tasks onto executors it already has.

Why does my driver run out of memory when the data fits in the cluster?

Because something pulled data back to the driver — collect(), toPandas(), or a broadcast of something too large. Executors hold the distributed data; the driver only holds the plan and whatever you explicitly bring back to it.

How many cores should an executor have?

Four to five is the usual sweet spot. Fewer wastes the JVM overhead per executor; more causes HDFS and object-store client contention, and makes garbage collection pauses affect more tasks at once. Prefer several medium executors over one enormous one.

Spark Architecture: Driver, Executors, and Cluster Managers

Spark’s API hides the cluster well. Understanding what sits behind it is what turns a mysterious out-of-memory error into an obvious one.

The three components

        ┌─────────────────────────────────────────────┐
        │  Driver (your program)                      │
        │  - builds the logical plan                  │
        │  - optimises it into a physical plan        │
        │  - splits stages into tasks                 │
        │  - schedules tasks onto executors           │
        └────────────────┬────────────────────────────┘
                         │  asks for resources
                         ▼
        ┌─────────────────────────────────────────────┐
        │  Cluster manager (YARN / K8s / standalone)  │
        │  - owns the machines, grants containers     │
        └────────────────┬────────────────────────────┘
                         │  launches
          ┌──────────────┼──────────────┐
          ▼              ▼              ▼
    ┌──────────┐   ┌──────────┐   ┌──────────┐
    │ Executor │   │ Executor │   │ Executor │
    │ - tasks  │   │ - tasks  │   │ - tasks  │
    │ - cache  │   │ - cache  │   │ - cache  │
    └──────────┘   └──────────┘   └──────────┘

The driver is the only part that runs your code’s control flow. Everything inside a transformation runs on executors.

Watching it start

spark-shell --master "local[4]" --driver-memory 2g

Setting default log level to "WARN".
Spark context Web UI available at http://192.168.1.24:4040
Spark context available as 'sc' (master = local[4], app id = local-1771155301842).
Spark session available as 'spark'.

Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.5.4
      /_/

Using Scala version 2.12.18 (OpenJDK 64-Bit Server VM, Java 17.0.13)

master = local[4] means the driver and four executor threads share one JVM. That is the same code path as a real cluster, with the network calls short-circuited.

Ask the running context what it has:

println(s"master:      ${sc.master}")
println(s"app id:      ${sc.applicationId}")
println(s"executors:   ${sc.getExecutorMemoryStatus.size}")
println(s"cores:       ${sc.defaultParallelism}")
println(s"deploy mode: ${sc.deployMode}")

master:      local[4]
app id:      local-1771155301842
executors:   1
cores:       4
deploy mode: client

One executor because local mode collapses them into the driver JVM; four cores because you asked for local[4].

Where your code runs

This is the distinction that matters most in practice:

val nums = sc.parallelize(1 to 10, 4)

println("this prints on the driver")

nums.foreach { n =>
  println(s"this prints on an executor: $n")   // goes to the executor's stdout
}

val collected = nums.collect()
println(s"back on the driver: ${collected.mkString(",")}")

this prints on the driver
this prints on an executor: 1
this prints on an executor: 2
...
back on the driver: 1,2,3,4,5,6,7,8,9,10

In local mode the executor output appears in your terminal because it is the same JVM. On a real cluster those lines land in each executor’s log on its own machine, which is why println debugging inside a transformation appears to do nothing.

The same applies to mutable state:

var counter = 0
nums.foreach(n => counter += n)
println(s"counter on driver: $counter")

counter on driver: 0

Zero, not 55. Each executor got a copy of counter, incremented its copy, and threw it away. This is the classic closure mistake. Use an accumulator:

val acc = sc.longAccumulator("total")
nums.foreach(n => acc.add(n))
println(s"accumulator: ${acc.value}")

accumulator: 55

Accumulators are the supported channel for executors to send values back to the driver.

Jobs, stages, tasks

val result = sc.parallelize(1 to 1000000, 8)
  .map(n => (n % 100, n))
  .reduceByKey(_ + _)
  .collect()

println(s"groups: ${result.length}")

groups: 100

That one action produced one job. Check the structure:

sc.statusTracker.getJobIdsForGroup().foreach { id =>
  val info = sc.statusTracker.getJobInfo(id).get
  println(s"job $id: ${info.stageIds().length} stages")
  info.stageIds().foreach { sid =>
    val s = sc.statusTracker.getStageInfo(sid).get
    println(s"  stage $sid: ${s.numTasks} tasks")
  }
}

job 0: 2 stages
  stage 0: 8 tasks
  stage 1: 8 tasks

Two stages because reduceByKey shuffles. Eight tasks each because the RDD has eight partitions. The hierarchy is always: action → job → stages (split by shuffles) → tasks (one per partition).

Resource shape

On a real cluster you declare what you want:

spark-submit \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 4g \
  --executor-memory 8g \
  --executor-cores 4 \
  --num-executors 10 \
  app.py

25/02/15 09:14:22 INFO Client: Application report for application_1771000000_0042 (state: ACCEPTED)
25/02/15 09:14:31 INFO Client: Application report for application_1771000000_0042 (state: RUNNING)
25/02/15 09:14:31 INFO Client:
	 client token: N/A
	 ApplicationMaster host: worker-07.internal
	 queue: analytics
	 start time: 1771146862104
	 tracking URL: http://rm.internal:8088/proxy/application_1771000000_0042/

Ten executors, four cores each — forty tasks in parallel. The total memory the cluster must grant is larger than 10 × 8g, because each executor also gets an overhead allocation:

executor container = executor-memory + max(384MB, 0.10 × executor-memory)
                   = 8192MB + 819MB = 9011MB

That overhead covers JVM internals, off-heap buffers, and the Python worker processes if you are using PySpark. Underestimating it is the usual cause of a container being killed by YARN:

Container killed by YARN for exceeding physical memory limits.
9.2 GB of 9.0 GB physical memory used.
Consider boosting spark.yarn.executor.memoryOverhead.

Client versus cluster mode

--deploy-mode client    # driver runs where you typed the command
--deploy-mode cluster   # driver runs on a cluster node

In client mode the driver is your laptop or an edge node — convenient for interactive work, and fatal for a long job if your SSH session drops. In cluster mode the driver lives inside the cluster and survives your terminal closing, which is what production scheduling needs.

The difference shows up in where output appears:

# client mode
$ spark-submit --deploy-mode client app.py
+-------+-----+
|country|count|
+-------+-----+
|UK     | 1284|
+-------+-----+

# cluster mode
$ spark-submit --deploy-mode cluster app.py
25/02/15 09:20:03 INFO Client: Application report ... (state: FINISHED)
$ yarn logs -applicationId application_1771000000_0043 | grep -A4 country

Nothing prints locally in cluster mode — your show() went to the driver’s log on a cluster node.

Configuration precedence

Three places to set the same option, and a fixed order of precedence:

// 1. In code — highest precedence
val spark = SparkSession.builder.config("spark.sql.shuffle.partitions", "50").getOrCreate()

# 2. On the command line
spark-submit --conf spark.sql.shuffle.partitions=100 app.py

# 3. In spark-defaults.conf — lowest
echo "spark.sql.shuffle.partitions 200" >> $SPARK_HOME/conf/spark-defaults.conf

Check what actually took effect:

println(spark.conf.get("spark.sql.shuffle.partitions"))

Code wins. This is worth remembering when a --conf flag appears to be ignored — something in the application is overriding it.

Practice

1. Increment a counter inside foreach and print it on the driver.

counter on driver: 0

Each executor mutated its own serialised copy. This is not a bug in Spark — it is what distributing a closure means. Use longAccumulator for counters, or restructure so the value comes back through reduce or collect.

2. Run with local[1] and local[8], timing the same job.

local[1]: 12.41s
local[8]:  2.18s

Not quite 8× — some stages are serial, and there is scheduling overhead per task. This is Amdahl’s law showing up in a Spark job, and it is why doubling executors rarely halves runtime.

3. Call collect() on a DataFrame larger than driver memory.

java.lang.OutOfMemoryError: Java heap space
	at org.apache.spark.sql.Dataset.collectFromPlan(Dataset.scala:3715)

The driver, not an executor. collect() pulls every row into the driver’s heap regardless of how much cluster memory you have. Use show(), limit().collect(), or write to storage.

4. Set the same config in code and on the command line with different values.

--conf spark.sql.shuffle.partitions=100  →  reported value: 50

The in-code value wins. Some settings, notably spark.driver.memory, cannot be set in code at all because the JVM has already started by then — those must come from spark-submit or the defaults file.

Next: the RDD API underneath the DataFrames, and the few places it still matters.