Spark Core · Module 3 · Deep dive
How plans become stages and tasks
The full submit sequence, what your Python and the cluster are each doing over time, the calls that break the "lazy" rule, and how the scheduler cuts a plan into stages. Start with the overview for the basics.
spark-submit to job completion, in ten steps
- Submit.
spark-submit(or a Databricks job or notebook) starts the driver with your code and configuration. - Driver starts. It creates the SparkSession and asks the cluster manager for executors.
- Executors launch and register with the driver.
- Your code runs on the driver. Each transformation adds a step to a logical plan. No data moves.
- An action is called. Spark creates a job for that DataFrame's lineage.
- Catalyst optimises the logical plan and picks a physical plan (Module 4).
- The DAG scheduler cuts the physical plan into stages at every shuffle and submits them in dependency order.
- The task scheduler sends one task per partition to free executor cores, preferring executors that already hold the data (locality).
- Executors run tasks. A stage that ends in a shuffle writes shuffle files to local disk; the next stage fetches them over the network.
- The job finishes when its last stage completes, and the result returns to Python. Failed tasks are retried (4 attempts by default) using lineage. The application ends when the SparkSession stops.
What Python and the cluster are doing over time
The same code as the overview, drawn as two lanes. Python is busy for microseconds per transformation, then blocks at each action while the cluster runs a job.
| Line | Python | Spark |
|---|---|---|
read | Runs | Lists files and reads the schema (may be a tiny job) |
filter | Runs instantly | The logical plan grows by one node |
count() | Blocks | Optimises read → filter, runs Job 1, returns a number |
if n > 0 | Plain Python | Not involved |
groupBy | Runs instantly | A new plan built on top of ae's lineage |
write | Blocks | Runs Job 2, which re-reads and re-filters orders |
- Why Spark can't look ahead: code after an action can depend on its result (
if n > 0). Spark can't plan what it doesn't know will run. - Each action re-runs its lineage. Here
ordersis read twice. Cacheaebefore the first action, or drop the extracount(). - Actions run one after another. Two independent writes are two sequential jobs, unless you start them from separate threads.
Not quite lazy: calls that run work immediately
| Call | What runs eagerly |
|---|---|
spark.read… with inferSchema=true (CSV, JSON) | A job that scans data to work out column types. Give a schema to avoid it. |
spark.read.parquet(…) on many files | File listing and footer reads to build the schema |
| DataFrame transformations in classic PySpark | Analysis: a misspelt column raises AnalysisException on that line, before any action |
df.checkpoint() | Runs a job to materialise the data (eager by default) |
RDD sortByKey | A sampling job to decide range-partition boundaries |
spark.sql("CREATE TABLE …"), INSERT, DDL | Commands execute when called |
df.cache() | Nothing: caching is lazy, and data is cached during the next action |
df.explain() | Planning only; no job runs |
Why lazy, and why immutable
- Whole-plan optimisation. Seeing the full chain lets Catalyst push filters into the file scan and read only the needed columns.
- Fewer passes. Consecutive narrow steps are fused into one stage, so each row is read once and flows through all of them.
- Fault tolerance through lineage. The plan is a recipe. If an executor dies, Spark recomputes only the lost partitions from that recipe; no data replication needed.
An RDD, and the DataFrame built on it, is an immutable, partitioned dataset plus the lineage that produced it. Every transformation returns a new dataset. Immutability is what makes recomputation safe (a rebuilt partition equals the original) and lets tasks run in parallel without locks.
Partitions, narrow and wide
map vs flatMap
rdd = sc.parallelize(["spark is fast", "delta lake"])
rdd.map(lambda s: s.split(" ")).collect()
# [['spark','is','fast'], ['delta','lake']] one output per input
rdd.flatMap(lambda s: s.split(" ")).collect()
# ['spark','is','fast','delta','lake'] zero or more outputs, flattenedThe DataFrame equivalent of flatMap is F.explode(F.split("text", " ")). Both are narrow.
How the scheduler cuts stages
- The DAG scheduler starts from the operation that produces the action's result and walks backwards through the plan's dependencies.
- Narrow dependencies are kept in the current stage. Each shuffle dependency ends the walk for that branch and creates a parent stage that must write its shuffle output first.
- The final stage is a ResultStage (it computes the action's result). Every stage that feeds a shuffle is a ShuffleMapStage.
- Stages are submitted parent-first. Parents that don't depend on each other, such as the two scans before a join, can run at the same time.
How many tasks
| Stage type | Tasks |
|---|---|
| Reading files | About one per spark.sql.files.maxPartitionBytes (128 MB) of input; small files are packed together |
| After a shuffle | spark.sql.shuffle.partitions (200), or fewer if AQE coalesces |
After repartition(n) / coalesce(n) | n |
RDD from parallelize | spark.default.parallelism (usually total cores) |
Skipped stages
In the Spark UI, a stage marked skipped was needed by this job, but its shuffle output already existed from an earlier job, so Spark reused it. That's normal and good: it's Spark avoiding recomputation, for example when two actions share the same aggregation.
write can appear as several jobs. Read them together in the SQL / DataFrame tab, which shows the whole query.reduceByKey vs groupByKey
Both group by key and both shuffle. The difference is how much they shuffle.
groupByKey
Ships every (key, value) pair across the network, then builds the full list of values per key in memory on the reducer side.
map side: (a,1) (a,1) (a,1) (b,1)
shuffled: 4 records
A hot key's whole value list must fit in one task's memory: OOM risk.
reduceByKey
Combines values on the map side first (like a Hadoop Combiner), then shuffles one partial result per key per partition.
map side: (a,3) (b,1)
shuffled: 2 records
Less network, less memory. Use it, or aggregateByKey, whenever the function is associative.
DataFrames do this automatically: groupBy().agg(sum(…)) plans a partial_sum before the shuffle and a final sum after. You can see both in explain(). Background: the MapReduce Combiner.
Compared with Lakeflow Declarative Pipelines
Plain PySpark is lazy per action. Lakeflow Declarative Pipelines (formerly Delta Live Tables) work the opposite way: your Python is evaluated once, up front, only to build a graph. Each table function returns a DataFrame definition but runs nothing. The pipeline engine then sees every table and dependency together, orders them, and runs the whole graph with retries, incremental processing and data-quality checks. That's why calling count() or collect() inside a pipeline table function is an anti-pattern.
A full comparison page will come with the Delta Lake & Databricks path.
Check yourself
What's the difference between a ShuffleMapStage and a ResultStage?
collect, a number for count, files for write).