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. In the first lab, constructing a filtered aggregation visits zero source partitions. Collecting its seven totals then visits all eight input partitions.

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 stage task counts printed by the lab are sorted for reproducibility, not ordered as an execution timeline.

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. The small lab uses StatusTracker immediately after completion; its retained, weakly consistent metadata is not 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. The lab rejects more than a declared number of rows after taking at most one extra row. That demonstrates a row bound, not protection 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. In the RDD lab, repartition(16) before reduceByKey(…, 4) adds a shuffle and a 16-task stage without changing any country total. That supports removing this unnecessary redistribution; repartition can still be justified to change distribution 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. In the cache lab, three counts return the same 100,000 rows. The first computes eight upstream partitions, the second reuses them, and after unpersist the third computes eight again. These accumulator observations describe this successful run; 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 tested version. 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. The lab explicitly requests MEMORY_ONLY, whose replication is one. 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.

The separate loss model removes the failed holder before deciding what to rebuild. Losing worker 0 removes four sole copies; placing each partition on both workers leaves a survivor and requires no source reread in that model. It represents a narrow filter over unchanged, available inputs, not a real distributed failure test. 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

Read the final lab’s 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. The initial plan in the lab is not final; after collect, the tested fixture shows a final adaptive plan and a coalesced shuffle read. This observed coalescing 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.

Lab: local execution and explicit models

Verified with Python 3.12.14, PySpark 4.2.0, and Java 17.0.20.1. Install pyspark==4.2.0 in a virtual environment and set JAVA_HOME to a compatible Java installation. Driver and Python workers must use the same Python minor version; the setup selects the current interpreter. These are current verification versions, not a claim about the 2018 work period. Run the setup before each solution in a standalone Python process without an existing Spark session. local[4] runs locally with four worker threads; it does not launch four independent cluster executors. It needs loopback communication. The setup disables the browser UI and uses short-lived temporary storage.

Spark startup messages about logging, incubator modules, and falling back from the native Hadoop library appeared in the tested environment. Python shuffle also advised installing psutil for better spill support. These stderr messages are not the recorded stdout below. The examples completed, but this small run does not validate large-data spill behavior. Exercises 3 and 6 are explicit arithmetic/state models; the other exercises execute Spark. The synthetic orders use integer cents in one assumed currency, seven country keys, and 200,000 rows.

import os, sys, math, tempfile
from contextlib import contextmanager, redirect_stdout
from io import StringIO
from collections import defaultdict
from operator import add
from pyspark import StorageLevel
from pyspark.sql import SparkSession, functions as F

N = 200_000
os.environ["PYSPARK_PYTHON"] = sys.executable
os.environ["SPARK_LOCAL_IP"] = "127.0.0.1"

@contextmanager
def local_spark(aqe=False):
    with tempfile.TemporaryDirectory(prefix="de020-") as work:
        spark = (SparkSession.builder.master("local[4]").appName("de020-execution")
                 .config("spark.driver.host", "127.0.0.1")
                 .config("spark.driver.bindAddress", "127.0.0.1")
                 .config("spark.ui.enabled", "false")
                 .config("spark.ui.showConsoleProgress", "false")
                 .config("spark.sql.warehouse.dir", work)
                 .config("spark.sql.shuffle.partitions", "8")
                 .config("spark.sql.adaptive.enabled", str(aqe).lower())
                 .getOrCreate())
        spark.sparkContext.setLogLevel("ERROR")
        try:
            yield spark
        finally:
            spark.stop()

def pair(i):
    return (str(i % 7), (i * 37 % 20000) + 500)

def reference_sums():
    totals = defaultdict(int)
    for i in range(1, N + 1):
        country, cents = pair(i)
        if cents > 1000:
            totals[country] += cents
    return dict(totals)

def stage_task_counts(sc, group):
    tracker = sc.statusTracker()
    stages = set()
    for job in tracker.getJobIdsForGroup(group):
        info = tracker.getJobInfo(job)
        if info is None:
            raise RuntimeError("job metadata not available; inspect Spark UI or event log")
        stages.update(info.stageIds)
    counts = []
    for stage in stages:
        info = tracker.getStageInfo(stage)
        if info is None:
            raise RuntimeError("stage metadata not available; inspect Spark UI or event log")
        counts.append(info.numTasks)
    return sorted(counts)

1. An action turns a dependency graph into work. Predict the partition visits before and after collecting the seven totals.

with local_spark() as spark:
    sc = spark.sparkContext
    sc.setJobGroup("lazy", "lazy RDD example")
    source_visits = sc.accumulator(0)
    def read_partition(rows):
        source_visits.add(1)
        yield from rows
    source = sc.parallelize(range(1, N + 1), 8).mapPartitions(read_partition)
    revenue = source.map(pair).filter(lambda row: row[1] > 1000).reduceByKey(add, numPartitions=4)
    print("before action, partition visits:", source_visits.value, "jobs:", len(sc.statusTracker().getJobIdsForGroup("lazy")))
    result = dict(revenue.collect())
    print("after action, partition visits:", source_visits.value)
    print("stage task counts:", stage_task_counts(sc, "lazy"))
    print("all country sums correct:", result == reference_sums())
# before action, partition visits: 0 jobs: 0
# after action, partition visits: 8
# stage task counts: [4, 8]
# all country sums correct: True

2. Compare the same result with an extra redistribution. Compare the baseline with repartition(16), keeping the filter and final aggregation identical.

with local_spark() as spark:
    sc = spark.sparkContext
    for label, extra in (("baseline", False), ("extra repartition", True)):
        sc.setJobGroup(label, label)
        rows = sc.parallelize(range(1, N + 1), 8).map(pair).filter(lambda row: row[1] > 1000)
        if extra:
            rows = rows.repartition(16)
        totals = dict(rows.reduceByKey(add, numPartitions=4).collect())
        print(label, "stage task counts:", stage_task_counts(sc, label), "same answer:", totals == reference_sums())
# baseline stage task counts: [4, 8] same answer: True
# extra repartition stage task counts: [4, 8, 16] same answer: True

3. Count ideal waves without inventing elapsed time. Use four slots and compare 2, 8, 16, and 400 equal-duration tasks.

def wave_budget(partitions, slots, assumed_task_ms=20):
    if (type(partitions) is not int or partitions < 0
            or type(slots) is not int or slots <= 0
            or type(assumed_task_ms) not in (int, float)
            or not math.isfinite(assumed_task_ms) or assumed_task_ms < 0):
        raise ValueError("invalid task/slot/overhead inputs")
    waves = (partitions + slots - 1) // slots
    last_busy = 0 if waves == 0 else partitions - (waves - 1) * slots
    return waves, last_busy, partitions * assumed_task_ms
for partitions in (2, 8, 16, 400):
    waves, busy, aggregate_ms = wave_budget(partitions, 4)
    print(partitions, "tasks; ideal waves:", waves, "last-wave busy slots:", busy,
          "hypothetical summed task overhead ms:", aggregate_ms)
# 2 tasks; ideal waves: 1 last-wave busy slots: 2 hypothetical summed task overhead ms: 40
# 8 tasks; ideal waves: 2 last-wave busy slots: 4 hypothetical summed task overhead ms: 160
# 16 tasks; ideal waves: 4 last-wave busy slots: 4 hypothetical summed task overhead ms: 320
# 400 tasks; ideal waves: 100 last-wave busy slots: 4 hypothetical summed task overhead ms: 8000

4. Bound a preview at the receiver. Count the full RDD, inspect five rows, and reject a result that exceeds a declared row limit.

def bounded_collect(rdd, max_rows):
    if type(max_rows) is not int or max_rows < 0:
        raise ValueError("max_rows must be a nonnegative integer")
    rows = rdd.take(max_rows + 1)
    if len(rows) > max_rows:
        raise ValueError("result exceeds declared row bound")
    return rows

with local_spark() as spark:
    values = spark.sparkContext.parallelize(range(N), 8)
    print("count:", values.count())
    print("bounded preview:", values.take(5))
    try:
        bounded_collect(values, 5)
    except ValueError as error:
        print("bounded collection rejected:", str(error))
    small = spark.sparkContext.parallelize([1, 2, 3], 2)
    print("small collected result:", bounded_collect(small, 3))
# count: 200000
# bounded preview: [0, 1, 2, 3, 4]
# bounded collection rejected: result exceeds declared row bound
# small collected result: [1, 2, 3]

5. Observe reuse and explicit cache removal. Count a persisted RDD twice, remove its cache, and count again.

with local_spark() as spark:
    sc = spark.sparkContext
    visits = sc.accumulator(0)
    def traced_partition(rows):
        visits.add(1)
        for i in rows:
            if i % 2 == 0:
                yield i
    cached = sc.parallelize(range(N), 8).mapPartitions(traced_partition).persist(StorageLevel.MEMORY_ONLY)
    print("after persist, partition visits:", visits.value)
    first = cached.count()
    after_first = visits.value
    second = cached.count()
    after_second = visits.value
    cached.unpersist(blocking=True)
    third = cached.count()
    print("counts:", first, second, third)
    print("new upstream partition visits:", after_first, after_second - after_first, visits.value - after_second)
    print("requested replication:", StorageLevel.MEMORY_ONLY.replication)
# after persist, partition visits: 0
# counts: 100000 100000 100000
# new upstream partition visits: 8 0 8
# requested replication: 1

6. Remove failed cache holders before rebuilding. Model loss of worker 0 with one and two cache copies.

def missing_after_loss(holders, failed):
    failed = set(failed)
    survivors = {p: set(workers) - failed for p, workers in holders.items()}
    missing = sorted(p for p, workers in survivors.items() if not workers)
    return survivors, missing

source_parts = [list(range(p * 10, (p + 1) * 10)) for p in range(8)]
expected = {p: [x for x in rows if x % 2 == 0] for p, rows in enumerate(source_parts)}
for replication in (1, 2):
    holders = {p: {p % 2} if replication == 1 else {0, 1} for p in range(8)}
    survivors, missing = missing_after_loss(holders, {0})
    rebuilt = {p: [x for x in source_parts[p] if x % 2 == 0] for p in missing}
    print("modeled cache copies:", replication, "lost partitions:", missing,
          "source partitions reread:", len(rebuilt), "rebuilt values correct:", all(rebuilt[p] == expected[p] for p in rebuilt))
# modeled cache copies: 1 lost partitions: [0, 2, 4, 6] source partitions reread: 4 rebuilt values correct: True
# modeled cache copies: 2 lost partitions: [] source partitions reread: 0 rebuilt values correct: True

7. Compare initial and final DataFrame plans. Run the same country aggregation through Spark SQL with AQE enabled.

with local_spark(aqe=True) as spark:
    orders = spark.range(1, N + 1, numPartitions=8).select(
        F.col("id").alias("order_id"), (F.col("id") % 7).cast("string").alias("country"),
        (F.col("id") * 37 % 20000 + 500).alias("amount_cents"))
    query = orders.filter(F.col("amount_cents") > 1000).groupBy("country").agg(F.sum("amount_cents").alias("revenue"))
    before = StringIO()
    with redirect_stdout(before):
        query.explain(mode="simple")
    result = {row.country: row.revenue for row in query.collect()}
    after = StringIO()
    with redirect_stdout(after):
        query.explain(mode="simple")
    initial, final = before.getvalue(), after.getvalue()
    print("initial plan has adaptive wrapper:", "AdaptiveSparkPlan" in initial)
    print("initial plan has partial sum and hash exchange:", "partial_sum" in initial and "Exchange hashpartitioning" in initial)
    print("completed adaptive plan:", "isFinalPlan=true" in final)
    print("observed coalesced shuffle read:", "AQEShuffleRead coalesced" in final)
    print("DataFrame sums match RDD reference:", result == reference_sums())
# initial plan has adaptive wrapper: True
# initial plan has partial sum and hash exchange: True
# completed adaptive plan: True
# observed coalesced shuffle read: True
# DataFrame sums match RDD reference: True