Apache Spark Architecture and Execution

What Spark coordinates

A country-level revenue report starts with many order partitions and ends with seven totals. Spark must assign the input work, bring contributions for each country together, and return or store the answer. Spark can pipeline compatible operations and reuse persisted results across actions; it does not require all intermediate data to remain in RAM. Shuffle files, spill, source reads, and final writes still matter.

This article follows batch execution from a lazy expression to tasks, then explains partition sizing, memory, and recovery. The examples use both RDDs and DataFrames so the scheduler and SQL optimizer remain distinct. Structured Streaming adds state, progress tracking, and recovery rules beyond this batch scope. Timing and failure behavior require runtime evidence as well as a plan.

Lazy transformations and two kinds of plan

An RDD (resilient distributed dataset) represents a collection divided into logical partitions and the dependencies needed to compute them. Its transformations create new RDDs rather than changing the original collection in place. This API immutability does not freeze an external source: recomputation still needs the intended input version. Calling map or filter constructs another RDD; an action such as count or collect requests computation. The dependency history is its lineage.

A DataFrame describes rows through a schema and a relational plan. Spark SQL resolves columns and types, optimizes the logical plan, and selects physical operators. Built-in expressions expose structure that can support predicate movement, column pruning, and suitable data-source pushdown. An arbitrary Python function inside RDD.map is opaque to that relational optimizer. Lazy evaluation alone does not make every user function optimizable. Schema inference, file discovery, and planning can also perform I/O before the final data action; “lazy” describes deferred computation, not a promise that every API call does nothing.

A narrow dependency lets a parent partition feed at most one child partition. Maps and filters are typical examples; coalescing without shuffle can combine several parents into a child. These dependencies can be pipelined in a task without regrouping data by key. Reading a remote source may still use the network. A shuffle dependency redistributes records across partitions, commonly by key. Map tasks produce buckets; downstream tasks fetch and aggregate their assigned buckets. Data may be read locally or remotely. Joins can reuse existing distribution or broadcast a small side: a broadcast exchange still transfers data even when the large side avoids a shuffle.

Jobs, stages, and task attempts

In the simple RDD example, eight source partitions are mapped and filtered, then reduceByKey produces four output partitions. The driver schedules a shuffle-map stage with eight tasks and a result stage with four tasks. Map and filter run inside the input tasks; adding an ordinary map does not add a separate stage, but its CPU and serialization costs remain.

The DAG scheduler separates work at shuffle dependencies and submits ready stages. The task scheduler places their tasks on available executors and handles attempts. A failed task can be retried; speculative execution can create another attempt for a slow task when configured. Thus twelve logical tasks do not always mean twelve attempts. Completed shuffle outputs or cached partitions can also avoid repeated upstream work.

“Shuffles plus one” describes this simple linear, uncached RDD example. It is not a general stage-count formula. Branches, reused stages, and SQL execution complicate it, and one high-level action can launch multiple jobs. A Spark SQL Exchange or AQE query stage is not automatically a one-to-one scheduler stage. Use the executed plan together with the Jobs and Stages views or an event log. StatusTracker exposes retained, weakly consistent metadata rather than a durable monitoring record.

Simple RDD execution (schematic; no cache or retries)
8 input partitions -> map -> filter -> shuffle write
                                        |
4 output partitions <- reduceByKey <- shuffle read
                                        |
                               collect 7 country totals

Driver, executors, and resource allocation

The driver owns the application context, builds plans, coordinates scheduling, and receives action results. Executors run tasks and hold cache and shuffle state. A cluster manager—Spark Standalone, YARN, or Kubernetes—allocates resources for the application. A node may host more than one executor. In client deployment the driver runs with the submitting client; in cluster deployment it runs within the cluster. Its placement affects connectivity and application lifetime. See the cluster overview for these roles.

The driver is one logical coordinator per classic application, not necessarily one operating-system process: PySpark involves Python and a JVM. Spark Connect separates the client from a remote driver. Adding executors does not enlarge the memory of the driver or client that receives collect or toPandas results. Large plans and scheduling metadata can also pressure the driver.

A Python function sent to executors carries serialized code and captured values. Appending to a captured list or changing a normal counter inside that function does not update the driver’s original variable. Return data through a transformation and aggregate it, or use an appropriate shared-variable mechanism with its documented limits. Python workers also use memory outside the executor JVM heap; a JVM heap setting alone is not the total process-memory budget.

Use a distributed write for a large result, count when only a count is needed, and take a small bounded preview when inspecting rows. Row count alone is not a memory budget: one row can contain a huge value, and serialization changes its footprint. A preview that takes one more row than a declared limit can detect a row-bound violation, but cannot protect against every out-of-memory condition. Broadcast construction also needs attention to the size held by its producing and receiving processes.

Partitions, slots, and the cost of an extra shuffle

A task processes a stage partition. A slot is a useful shorthand for capacity to run one task concurrently. CPU-only capacity is roughly the sum of executor cores divided by CPUs requested per task, rounded down per executor; GPUs, resource profiles, competing work, and dynamic allocation can change it. With four available slots and equal-duration tasks, eight partitions need two ideal waves. Two partitions leave capacity unused. Four hundred tiny partitions need one hundred waves and incur task setup repeatedly.

The wave calculation is an idealized scheduling count, not a runtime estimator. Its assumed 20 ms per task gives 8,000 ms of aggregate overhead for 400 tasks, not a prediction of eight seconds elapsed. Uneven task duration can leave a long tail. Too few large partitions can also spill or exhaust memory. Start from data size and available resources, then measure task duration, memory pressure, and imbalance; a fixed tasks-per-core ratio cannot choose every workload’s partition count.

Input partitions depend on the source, splitability, file sizes, and reader settings. Small files can be grouped and large splittable files divided, so one file or HDFS block need not equal one Spark task. RDD defaultParallelism and SQL shuffle partition settings have different roles. SQL’s configured starting shuffle count can be changed by AQE at runtime. For the simple RDD aggregation above, repartition(16) before reduceByKey(…, 4) introduces another redistribution without changing the intended country totals. Remove redistribution that serves no required downstream distribution; repartition can still be justified for later work.

Caching does not preserve a lost executor’s only copy

persist selects a storage policy; it does not immediately fill the cache. An action materializes the partitions it needs. Reusing them can avoid upstream computation, while a partial action need not populate every partition. Transformation-side accumulator updates are not an exactly-once accounting mechanism under retries.

RDD cache defaults to MEMORY_ONLY; DataFrame cache uses MEMORY_AND_DISK_DESER in the version scoped here, PySpark 4.2.0. Execution memory for joins and aggregations competes with storage needs, and caching everything can increase eviction, spill, and garbage collection. Choose data that will be reused and release it when done. MEMORY_ONLY has one cache copy; local mode has no second executor on another machine to protect that copy. See the RDD persistence API and storage levels.

If an executor disappears, its only cached copy disappears too. Spark can recompute a missing partition from available lineage and inputs; a surviving cached ancestor, shuffle output, or replicated cache copy can reduce that work. Lost shuffle files can require upstream recomputation beyond one result partition. Shuffle services and decommissioning arrangements affect what survives. Lineage is a recipe, not a backup of unavailable source data: mutable inputs or nondeterministic functions can change recomputed results. Retried external side effects need their own idempotency design.

Reliable checkpointing can truncate lineage by storing a durable recovery boundary; local checkpointing sacrifices that durability. Driver failure generally ends the application unless an external recovery arrangement restarts it. Executor lineage recovery alone does not restore a lost driver.

Read the executed SQL plan and its measurements

Consider an illustrative SQL aggregation from the source upward: Range produces eight input partitions, Filter selects amounts, Project keeps expressions needed downstream, and partial_sum combines contributions locally. Exchange hashpartitioning groups those partial contributions by country, then the final HashAggregate completes the sums. Partial aggregation can make the shuffled data much smaller than the source. An Exchange identifies redistribution, not a measured byte count.

The example uses generated Range rows, so its Filter is not evidence of file-reader predicate pushdown. For a file scan, inspect the relevant scan fields such as ReadSchema, PartitionFilters, and PushedFilters, and distinguish delegated predicates from residual filtering. Operator identifiers and whole-stage code-generation markers are not scheduler stage IDs.

With adaptive query execution enabled, Spark can revise selected parts of the physical plan using runtime statistics, for example coalescing small shuffle partitions or changing a join strategy. A coalesced shuffle read is not guaranteed for every dataset or configuration. Compare initial and final plans, then inspect actual task counts. See SQL performance tuning and AQE.

A long task tail suggests checking key skew, input sizes, spill, garbage collection, and slow hosts. High shuffle volume can be appropriate for a legitimate regrouping; spill can be a normal memory tradeoff. Neither symptom alone identifies the cause. Compare task duration distributions and input, shuffle, spill, and GC metrics in the Spark UI, using the same data and result when comparing changes. A plan reveals avoidable work but cannot predict most runtime behavior before execution.