What Actually Happens When You Submit a Spark Job

Most Spark tutorials teach you the API. Filter this, join that, write the result. But when something breaks at 3am and you're staring at a failed stage with cryptic errors, the API doesn't help. You need to know what's actually going on inside.

I've debugged enough Spark jobs to know that understanding the lifecycle is the difference between guessing and knowing. So here's the full picture from hitting enter on spark-submit to either success or a very educational failure.

The Job

We're tracing through this:

df = spark.read.parquet("s3://data/trips")

result = (
    df.filter("trip_date >= '2026-01-01'")
      .join(users_df, "user_id")
      .groupBy("city")
      .agg({"fare": "sum"})
)

result.write.mode("overwrite").parquet("s3://output")

Running on YARN. Kubernetes is the same story with different container management.

The Big Picture

A Spark application moves through these phases:

Submission → Driver Init → Executor Allocation → Planning → 
DAG Creation → Task Scheduling → Execution → Shuffle → Write → Cleanup

When something breaks, knowing which phase failed tells you where to look. Most of debugging is just figuring out which part of this pipeline went wrong.

Submission

You run:

spark-submit \
  --driver-memory 6g \
  --executor-memory 12g \
  --executor-cores 4 \
  --num-executors 40 \
  job.py

Two deploy modes matter here:

Client mode; the driver runs on your machine. Fine for debugging, terrible for production. Close your laptop and the job dies.

Cluster mode; the driver runs inside the cluster. This is how production works.

Either way, spark-submit bundles your code and config, talks to the cluster manager (YARN's ResourceManager, K8s API server, whatever), and says "start a driver somewhere." The cluster manager finds a spot, launches the Driver JVM, and now your application exists.

Driver Initialization

The Driver is the brain. When it boots up, it creates all the internal machinery:

Driver JVM
├── SparkSession / SparkContext
├── DAGScheduler (chops jobs into stages)
├── TaskScheduler (assigns tasks to executors)
├── BlockManagerMaster (tracks where data lives)
└── MapOutputTracker (tracks shuffle file locations)

Nothing has actually executed yet. Your code runs top-to-bottom, but Spark operations are lazy. That df.filter().join().groupBy() chain just builds a description of what you want. No data moves. The Driver is sitting there waiting for you to actually ask for results.

Executor Allocation

The Driver needs workers.

It tells the cluster manager: "I want 40 executors, 4 cores and 12GB each." The cluster manager scrounges up resources and starts launching containers. Each container becomes an Executor — a separate JVM that'll do the real data processing.

As each Executor starts, it phones home: "I'm alive, I have 4 cores, ready for work."

Driver
  ↕ (heartbeats, task assignments)
Executor 1  Executor 2  ...  Executor 40

Each Executor sets up its own:

• Memory Manager (carves up the heap)
• BlockManager (stores data, talks to other executors)
• Shuffle Manager (handles shuffle I/O)
• Thread pool (one thread per core)

Now you have 40 workers sitting idle, waiting.

With dynamic allocation, this works differently — executors spin up as needed instead of all at once. But the handshake is the same.

Query Planning (Catalyst)

Your code finally hits an action .write(). Actions trigger computation. Everything before was just building a plan.

Before going further, the terminology: one action creates one job. Jobs get split into stages at shuffle boundaries. Stages contain tasks one per partition. So our single .write() creates one job with maybe 3-4 stages, each with hundreds or thousands of tasks.

Catalyst takes your logical operations and figures out the smartest way to run them.

First, it builds a logical plan basically a tree of what you asked for:

Write
└── Aggregate (groupBy city, sum fare)
    └── Join (user_id)
        ├── Filter (trip_date >= 2026-01-01)
        │   └── Scan (trips parquet)
        └── Scan (users)

Then it optimizes. Catalyst has dozens of rules:

• Predicate pushdown: push that filter into the parquet read so you never load 2025 data
• Column pruning: you only need user_id, city, fare, so skip the other 50 columns
• Join reordering: if one table is tiny, maybe join it first
• Constant folding: pre-compute anything that doesn't need data

The optimized plan can look very different from what you wrote.

Next, physical planning. Spark picks actual algorithms broadcast join vs sort-merge, hash aggregation vs sort-based, where to put shuffle operators.

Finally, code generation. This is where Spark gets fast. Instead of interpreting the plan row-by-row, it generates optimized Java bytecode that fuses multiple operators into tight loops. "Whole-stage code generation" you can spot it in the Spark UI because those stages have a * next to them. If a stage doesn't have the asterisk, something blocked optimization (usually a UDF).

DAG and Stage Creation

The DAGScheduler takes the physical plan and cuts it into stages.

The rule: every shuffle creates a new stage. And shuffles only happen during wide transformations operations where data needs to move between partitions.

Wide transformations (cause shuffle, create new stage):

• groupBy, groupByKey, reduceByKey — data with same key must land on same partition
• join, cogroup — matching keys from both sides need to meet
• distinct — duplicates across partitions must be found
• repartition — explicitly redistributes data
• sortBy, orderBy — global ordering requires all data to coordinate

Narrow transformations (no shuffle, same stage):

• map, flatMap, filter — each input partition produces one output partition
• mapPartitions — operates within partition boundaries
• union — just combines partitions, no data movement
• coalesce (without shuffle) — merges partitions locally

The distinction matters for performance. Narrow transformations chain together in a single stage Spark pipelines them efficiently. Wide transformations force a stage boundary because the data has to physically move before the next operation can proceed.

Why the boundary? Shuffles are synchronization points. All map-side tasks have to finish before reduce-side tasks can start fetching. You can't overlap them.

For our job:

Stage 0: Read users parquet
Stage 1: Read trips parquet → Filter
Stage 2: Shuffle read from both → Join → Shuffle write for groupBy
Stage 3: Shuffle read → Aggregate → Write output

The DAG captures dependencies:

Stage 0 (users) ──┐
                  ├──→ Stage 2 (join) ──→ Stage 3 (agg + write)
Stage 1 (trips) ──┘

Stage 0 and Stage 1 run in parallel they don't depend on each other. Stage 2 waits for both. Stage 3 waits for Stage 2.

This parallelism matters. If trips is huge and users is small, reading users finishes while trips is still scanning. Spark doesn't waste time waiting.

Task Scheduling

The TaskScheduler matches tasks to executors.

For each stage, it creates a TaskSet all the tasks for that stage. Then it starts handing out tasks to available executor slots.

Locality matters. Spark prefers running tasks where the data already is:

1. PROCESS_LOCAL: data is in executor memory. Best case.
2. NODE_LOCAL: data is on the same machine, different executor or disk.
3. RACK_LOCAL: same rack, one network hop.
4. ANY: data is wherever. Full network transfer.

Spark waits a bit for better locality before giving up. spark.locality.wait controls how long.

As tasks finish, slots free up, more tasks get assigned. This continues until the stage is done.

One more thing: speculative execution. Sometimes a task runs slow bad disk, noisy neighbor, GC storm. If spark.speculation=true, Spark notices when a task takes way longer than its siblings and launches a backup copy on another executor. Whichever finishes first wins. Helps with stragglers, but wastes resources if your task durations naturally vary a lot.

Execution

Inside an Executor, actual work happens.

Each Executor has a thread pool matching its core count (4 threads in our case). Each thread runs one task at a time.

When a task arrives:

1. Deserialize the task code
2. Read input either from storage (first stage) or shuffle files (later stages)
3. Process records through the transformations
4. Write output shuffle files for intermediate stages, final output for the last one

Memory layout during execution:

Executor Heap (12 GB)
├── Reserved (300 MB)
└── Usable (11.7 GB)
    ├── Spark Memory (7 GB)
    │   ├── Execution: hash tables, sort buffers, join structures
    │   └── Storage: cached RDDs, broadcast variables
    └── User Memory (4.7 GB): your UDFs, custom objects

Tasks share the execution memory pool. One task building a huge hash table competes with the others in the same executor.

When memory runs low, Spark spills to disk. Slow, but better than OOM. You'll see "spill" in the Spark UI when this happens.

What to watch in the UI: Duration (huge variance = skew), GC Time (high = memory pressure), Shuffle Read/Write (unexpectedly large = join explosion), Spill (any = memory too tight). If 99 tasks finish in 10 seconds and one takes 10 minutes, you've got skew.

The Shuffle

Shuffles are where distributed computing gets expensive.

On the map side, as tasks process records, they partition output by the shuffle key (user_id for our join). Records buffer in memory, get sorted when the buffer fills, spill to disk, and merge into final shuffle files when the task completes. It's incremental, not a bulk write at the end.

On the reduce side, each task needs data from every map task. It asks the MapOutputTracker where the blocks are, fetches them from all the executors that have relevant data, merges them, processes the result.

Lots of network traffic. 40 executors, 1000 shuffle partitions — that's potentially 40,000 block fetches.

What breaks:

• Executor dies → shuffle files lost → stage retries
• Network congestion → fetch timeouts → FetchFailedException
• Skewed partition → one task takes forever
• Too many tiny partitions → scheduling overhead dominates

External shuffle service helps with the first one. With spark.shuffle.service.enabled=true, a separate daemon serves shuffle files. Executor death doesn't lose data.

About retries: tasks retry up to 4 times by default (spark.task.maxFailures). If a task keeps failing, the stage fails. If shuffle data is lost, the previous stage re-runs to regenerate it. This is why one flaky executor can cause massive recomputation — everything that depended on its shuffle output has to redo work.

Adaptive Query Execution

With AQE enabled (Spark 3.x, and you should use it), the plan can change mid-flight.

After a stage completes, Spark knows actual shuffle sizes. It can combine tiny partitions to reduce overhead, split huge partitions so multiple tasks can handle them, or switch join strategies if one side turns out smaller than expected.

Runtime optimization based on real data instead of guesses. One of the best improvements in recent Spark versions.

Writing Output

The final stage computes aggregates and writes parquet files.

Each task writes its own file(s) to a temporary location. The output committer protocol handles atomicity driver verifies all tasks succeeded, then moves files from temp to final location. If anything failed, temp files get cleaned up.

Those _temporary directories you sometimes see? That's the commit protocol in action.

Cleanup

All stages done. Job succeeded.

The driver marks the application complete, executors get shutdown signals, containers return to the cluster manager, and Spark UI data moves to the history server if you have one configured.

Application's gone, but you can still dig through the history server to see what happened.

Component Reference

For debugging, it helps to know what lives where:

Driver side: SparkContext (owns everything), DAGScheduler (stages), TaskScheduler (task assignment), SchedulerBackend (talks to cluster manager), BlockManagerMaster (tracks blocks across cluster), MapOutputTracker (shuffle locations), Catalyst (optimization).

Executor side: Executor Backend (receives tasks, reports status), BlockManager (stores and serves data), Shuffle Manager (shuffle I/O), Memory Manager (execution/storage pools), task threads (your code runs here).

Cluster Manager: resource allocation, container lifecycle, node health, queue management (YARN) or pod scheduling (K8s).

When Things Break

Driver OOM deserves its own list. Common causes:

• .collect() on large data — pulls everything to driver memory
• Too many partitions — shuffle metadata scales with partition count, 100k+ can kill you
• Large broadcasts — driver holds the original before distributing
• Accumulator abuse — thousands of accumulators with big values
• Plan explosion — deeply nested or unioned queries

Full Timeline

spark-submit
    ↓
Cluster manager allocates driver container
    ↓
Driver boots, SparkContext initializes
    ↓
Driver requests executors
    ↓
Cluster manager launches executor containers
    ↓
Executors register with driver
    ↓
Your code runs, builds lazy plan
    ↓
.write() called → Job 0 created
    ↓
Catalyst optimizes
    ↓
DAGScheduler creates stages
    ↓
TaskScheduler assigns Stage 0 + Stage 1 tasks (parallel)
    ↓
Executors run, write shuffle files
    ↓
Both stages complete, Stage 2 scheduled
    ↓
Executors fetch shuffle, join, write more shuffle
    ↓
Stage 2 complete, Stage 3 scheduled
    ↓
Executors fetch, aggregate, write output
    ↓
Job 0 succeeded
    ↓
Cleanup, executors shut down

How to Think About It

Spark application as a factory:

• Cluster Manager — the landlord, provides the building
• Driver — the manager, plans work, assigns tasks, tracks progress
• Executors — the workers, do the processing
• Shuffle — the conveyor system, moves data between stages
• Memory — the workspace, determines how much each worker handles

When jobs fail, you're debugging one of these: not enough building (resources), bad planning (optimizer choices), overwhelmed workers (memory/sizing), jammed conveyor (shuffle/network).