🔥 Spark & Databricks · Chapter 2

Spark's execution model: driver, executors, stages

Your cluster has 40 cores and your stage has 4 partitions. Thirty-six cores are being paid for to do nothing.

Study this lesson free in the app →

The basics

Spark's execution model is the division of labour inside a Spark cluster: one driver plans the work, many executors do it, and the unit of work they hand around is a task.

Why it exists: one machine cannot hold or scan a terabyte of rides. Spark splits the data into partitions and runs the same code on each partition in parallel, on different machines. Somebody has to decide who processes what, and somebody has to actually process it — those are the driver and the executors.

Mental model: a restaurant. The head chef (driver) reads the order, writes the plan, and hands out tickets. Line cooks (executors) each have a few burners (cores). One ticket (task) = one dish from one tray of ingredients (partition). The number of dishes cooking at once is the number of burners, not the number of cooks.

Explained

Meghalaya Cabs runs its nightly fare rollup on a 10-node cluster. The Spark UI shows a stage with 4 tasks — because the input was 4 big gzip files, and gzip cannot be split.

Four cores busy. Thirty-six idle. The job takes 50 minutes and the bill counts all 40 cores.

Nothing was wrong with the code. Somebody did not know how work maps to cores.

The wedding kitchen

A head chef gets an order for 2,000 plates. She does not cook. She reads the menu, breaks it into steps, and writes tickets: one ticket per tray of ingredients.

She has 10 line cooks, each with 4 burners. Forty dishes can cook at once. If she writes 400 tickets, the kitchen works through them in 10 rounds.

If she writes only 4 tickets — four enormous trays — 36 burners stay cold, however many cooks she hires.

And if one cook's station catches fire, she rewrites his tickets for someone else. If the head chef walks out, the whole order is lost.

One action, end to end

The driver is the only process that knows the whole plan. Executors only ever see one task at a time per core.

Who does what

Driver — runs main() / your notebook, holds the SparkSession, builds the DAG, splits it into stages, schedules tasks, tracks which executor holds which shuffle output, and receives anything you collect().

Executors — long-lived JVMs started for your application. Each has spark.executor.cores cores and spark.executor.memory of heap. They run tasks, write shuffle files to local disk, and hold cached partitions.

Cluster manager — allocates containers. It knows nothing about DataFrames.

Task — the smallest unit: one stage's code applied to one partition. A task runs on one core.

==Parallelism = min(partitions, executors × cores).== Too few partitions wastes cores; too many tiny ones wastes time on scheduling overhead (each task costs a few ms to launch).

Inside one executor (spark.executor.memory = 16g)

 container  ≈ 16g heap + 1.6g overhead (max(384MB, 10%))
 ┌──────────────────────────────────────────────┐
 │ reserved            300 MB                   │
 ├──────────────────────────────────────────────┤
 │ unified  = 0.6 × (16384 − 300) ≈ 9.6 GB      │
 │  ┌─────────────────────┬──────────────────┐  │
 │  │ storage (cache)     │ execution        │  │
 │  │ ~50% protected      │ joins, sorts,    │  │
 │  │ storageFraction=0.5 │ aggregations     │  │
 │  └─────────────────────┴──────────────────┘  │
 │      boundary moves: execution can evict     │
 │      cache above the protected half          │
 ├──────────────────────────────────────────────┤
 │ user memory  0.4 × 16084 ≈ 6.4 GB            │
 │ (your objects, UDF data structures)          │
 └──────────────────────────────────────────────┘
 4 cores → ~2.4 GB of unified memory per running task

Sizing the Meghalaya Cabs cluster

spark = (SparkSession.builder
  .appName('mc_fare_rollup')
  .config('spark.executor.instances', 10)
  .config('spark.executor.cores', 4)        # 4 tasks at once per executor
  .config('spark.executor.memory', '16g')   # JVM heap
  .config('spark.executor.memoryOverhead', '2g')  # off-heap, Python workers
  .config('spark.driver.memory', '8g')
  .getOrCreate())

rides = spark.read.parquet('s3://mc-lake/silver/rides/')
print(rides.rdd.getNumPartitions())   # 320 → 320 tasks in the scan stage
# 10 executors × 4 cores = 40 slots → 320 / 40 = 8 waves

print(spark.sparkContext.defaultParallelism)  # usually = total cores = 40

Rule of thumb most teams use: 4–5 cores per executor. One core per executor wastes JVM overhead; 16 cores per executor gives each task a thin slice of memory and garbage-collection pauses that stall all 16.

What changes at 100×

Waves matter more than cores. 4,001 tasks on 4,000 slots = 2 waves, the second with one task. Aim for task counts a little under a multiple of slots, or enough tasks (3–4× slots) that the tail is small.

The driver becomes the bottleneck with huge plans or many files: listing 2 million small files, broadcasting a 1GB table, or collect()ing results all happen on one JVM. Raise driver memory or, better, stop sending it data.

Dynamic allocation adds and removes executors with load — good for shared clusters, but executors that hold shuffle files can only be removed safely with an external shuffle service or shuffle tracking.

Executor loss is routine at scale. Spot instances get reclaimed; Spark recomputes the lost partitions from lineage. Driver loss is not routine — it ends the application.

Gotchas

• Fewer partitions than cores — a 4-partition stage on 40 cores runs at 10% utilisation. repartition after reading unsplittable files
• collect() / toPandas() on big data — all rows travel to the driver; OutOfMemoryError on the driver kills the whole job
• Fat executors — 16 cores sharing one heap means each task gets ~1/16 of execution memory and long GC pauses
• Forgetting overhead — PySpark UDFs run in Python worker processes outside the JVM heap; the container gets killed by YARN/K8s for exceeding its memory limit even though the heap was fine
• Thinking executors talk to each other through the driver — shuffle data goes executor-to-executor directly; the driver only tracks where it lives

Words to know

In two minutes

• The driver runs your code, turns DataFrame operations into a plan of stages, and schedules tasks; it does not touch the bulk of the data.
• Executors are JVM processes on worker nodes; each has a fixed number of cores and a fixed heap, and lives for the whole application.
• One task = one partition in one stage. A stage with 400 partitions has 400 tasks.
• Parallelism is capped by slots = executors × cores per executor. 10 executors × 4 cores = 40 tasks at a time; 400 tasks run in 10 waves.
• Executor heap minus 300MB reserved: 60% is unified memory (execution + storage, sharing a boundary), 40% is user memory. Overhead (default max(384MB, 10%)) sits outside the heap.
• Memory per task is roughly unified memory ÷ cores — more cores per executor means less memory per task.
• The cluster manager (YARN, Kubernetes, Standalone, or Databricks' own) only hands out containers; it never sees your rows.

Test yourself

There are 9 questions, 3 puzzles and 5 recall cards for this lesson in the app — the recall cards come back on later days, just before you would forget.