How Apache Spark Works Internally: Driver Architecture, DAG Scheduling, Shuffle Mechanics, and the Catalyst Optimizer Behind Distributed Data Processing

Most engineers who use Apache Spark can write a DataFrame transformation and submit a job. Fewer understand why a query that looks correct on paper runs for three hours and then OOMs. The gap is almost always in the engine: how Spark translates a logical plan into physical stages, how data moves between those stages, and how the optimizer decides what to rewrite. Understanding these mechanisms turns tuning from guesswork into deliberate configuration.

This article covers the components that matter for production systems: the driver-executor model, RDD lineage and lazy evaluation, DAG scheduling and stage boundaries, the sort-based shuffle implementation, Catalyst’s optimization passes, Tungsten’s off-heap memory model, and adaptive query execution. Each section explains the mechanism and then the operational implication.

The Driver-Executor Architecture

Spark separates coordination from computation. The driver runs the user’s main program and hosts the SparkContext, which is the entry point for all cluster operations. The executors are JVM processes launched on worker nodes that actually run tasks and store cached data.

When a job starts, SparkContext connects to a cluster manager (YARN, Kubernetes, or Spark standalone) and requests executor processes. Each executor registers back with the driver. From that point, the driver acts as a centralized scheduler: it breaks work into tasks, assigns them to executors, and tracks progress through a heartbeat protocol.

The driver holds the complete job graph in memory. This is both a strength and a failure mode. A driver with insufficient heap running a job that builds a large broadcast variable or collects results back from executors will OOM. The cluster manager sees healthy executors while the driver silently dies.

A key concept is the SparkContext lifecycle. Creating a second SparkContext in the same JVM used to cause errors (and still does without spark.driver.allowMultipleContexts). In practice, Spark 3.x manages this better, but the constraint reflects the fact that a single driver owns the task scheduler, the block manager, and the broadcast variable registry.

// TypeScript pseudocode: how the driver tracks executor state
interface ExecutorBackend {
  executorId: string;
  host: string;
  cores: number;
  memoryMB: number;
  status: "alive" | "lost" | "decommissioning";
  lastHeartbeat: number;
}

class TaskScheduler {
  private executors = new Map<string, ExecutorBackend>();
  private pendingTasks: TaskDescription[] = [];

  registerExecutor(backend: ExecutorBackend): void {
    this.executors.set(backend.executorId, backend);
    this.scheduleAvailableTasks();
  }

  private scheduleAvailableTasks(): void {
    for (const [id, executor] of this.executors) {
      if (executor.status === "alive" && this.pendingTasks.length > 0) {
        const task = this.pendingTasks.shift()!;
        this.launch(executor, task);
      }
    }
  }

  heartbeatReceived(executorId: string): void {
    const executor = this.executors.get(executorId);
    if (executor) executor.lastHeartbeat = Date.now();
  }
}

The practical implication: right-size the driver. For jobs with large collect operations or wide broadcast joins, allocate 8–16 GB to the driver. For streaming jobs where the driver runs continuously, treat driver memory as a first-class resource alongside executor memory.

RDD Lineage and Lazy Evaluation

The foundational abstraction in Spark is the Resilient Distributed Dataset (RDD). An RDD is not a collection of data. It is a description of how to compute data: which parent RDDs it depends on, which partitions those map to, and which function transforms them.

This matters because Spark evaluates lazily. When you write val result = rdd.filter(...).map(...).groupBy(...), nothing executes. Spark builds a lineage graph. Execution only begins when an action is called: count, collect, save, take. At that point, the DAG scheduler walks the lineage graph and plans execution.

Lineage enables two things. First, fault recovery without replication: if an executor loses a partition, Spark recomputes it from the lineage rather than reading from a replica. Second, it enables the optimizer to see the full transformation chain before generating a physical plan.

The distinction between narrow and wide dependencies is central to how the DAG scheduler works. A narrow dependency means each output partition depends on at most one input partition (map, filter, union with aligned partitions). A wide dependency means one output partition depends on multiple input partitions (groupByKey, reduceByKey, join). Wide dependencies require a shuffle: data must move across the network before the next computation can proceed.

// TypeScript pseudocode: RDD dependency types
type Dependency = NarrowDependency | ShuffleDependency;

interface NarrowDependency {
  kind: "narrow";
  parentRdd: RDDId;
  // output partition P depends only on parent partition P
  partitionMapping: (outputPartition: number) => number[];
}

interface ShuffleDependency {
  kind: "shuffle";
  shuffleId: ShuffleId;
  partitioner: Partitioner; // determines which reducer gets each record
  serializer: Serializer;
  keyOrdering?: Ordering;
  aggregator?: Aggregator;
}

The takeaway: every wide dependency in your job graph is a shuffle boundary and a stage boundary. Minimizing unnecessary shuffles is the single highest-leverage performance optimization in Spark. Replacing a groupByKey (which materializes all values per key) with reduceByKey or aggregateByKey (which pre-aggregates on the map side) can cut shuffle data volume by orders of magnitude.

DAG Scheduling: From Logical Plan to Physical Stages

The DAGScheduler converts the RDD lineage into a set of stages separated at shuffle boundaries. A stage is a set of tasks that can all run without network communication: each task processes one partition using only narrow transformations.

The scheduler performs a reverse DFS from the final RDD. When it encounters a shuffle dependency, it creates a stage boundary: the upstream computation becomes a ShuffleMapStage (each task writes shuffle output to local disk), and the downstream computation becomes a ResultStage or another ShuffleMapStage.

Input RDDs -> [Stage 0: ShuffleMapStage] -> shuffle write
                                             |
                                        shuffle read
                                             |
                              [Stage 1: ShuffleMapStage] -> shuffle write
                                                             |
                                                        shuffle read
                                                             |
                                            [Stage 2: ResultStage]

Each stage executes as a set of tasks, one per partition. The DAGScheduler submits stages to the TaskScheduler, which handles the actual placement of tasks onto executors. The TaskScheduler uses a delay scheduling heuristic: it waits briefly for data-local tasks (tasks that can run on the executor holding the relevant HDFS block) before falling back to rack-local or any-node placement.

Stage retries happen independently. If a single task in a stage fails, only that task is retried, not the whole stage. If a shuffle map stage loses its output (executor dies after stage completion), the entire upstream stage must re-run because shuffle files live on local disk, not in a replicated store.

This is the reason executor decommissioning order matters. In Kubernetes, a spot instance preemption that kills an executor after a shuffle write but before the downstream stage completes forces a full upstream re-run. Enabling spark.shuffle.service.enabled with an external shuffle service decouples shuffle file lifetime from executor lifetime and eliminates this failure mode.

The Shuffle Architecture

Every shuffle in Spark passes through a ShuffleManager. The default since Spark 2.0 is the SortShuffleManager, which replaced the older HashShuffleManager. Understanding why it replaced it explains the performance model.

The old HashShuffleManager created one output file per (mapper, reducer) pair. A job with 1,000 map tasks writing to 2,000 reducers created 2 million files. This hammered the filesystem with random writes and caused metadata pressure on HDFS NameNodes.

SortShuffleManager instead sorts records by their target partition ID on the map side, writes a single sorted output file per map task, and writes an index file that records where each reducer’s data starts and ends. Reducers read contiguous byte ranges from the single file.

// TypeScript pseudocode: sort shuffle write path
interface ShuffleBlock {
  partitionId: number;  // target reducer partition
  key: Uint8Array;
  value: Uint8Array;
}

class SortShuffleWriter {
  private buffer: ShuffleBlock[] = [];
  private readonly spillThreshold = 64 * 1024 * 1024; // 64MB default

  insert(block: ShuffleBlock): void {
    this.buffer.push(block);
    if (this.bufferBytes() > this.spillThreshold) {
      this.spillSortedToDisk();
    }
  }

  commit(): { dataFile: string; indexFile: string } {
    // Final spill and merge all on-disk spill files
    this.spillSortedToDisk();
    return this.mergeSpillsAndWriteIndex();
  }

  private spillSortedToDisk(): void {
    // Sort in-memory buffer by partitionId (then by key if ordering required)
    this.buffer.sort((a, b) => a.partitionId - b.partitionId);
    this.writeToSpillFile(this.buffer);
    this.buffer = [];
  }

  private mergeSpillsAndWriteIndex(): { dataFile: string; indexFile: string } {
    // K-way merge of sorted spill files; write partition byte-range index
    return mergeSpillFiles(this.spillFiles);
  }
}

The sort can become a bottleneck when key cardinality is very high or when values are large. For cases where no sorting is needed (only partition assignment matters), the bypass merge threshold applies: if the number of reducers is below spark.shuffle.sort.bypassMergeThreshold (default 200), Spark skips the sort and writes one file per reducer bucket, then concatenates. This reverts to hash-shuffle behavior but at small scale where the file count is manageable.

On the read side, each reducer issues fetch requests to the executor block managers that wrote the shuffle data. spark.reducer.maxSizeInFlight (default 48 MB) controls how much data a reducer fetches at once. Increasing this for network-rich environments (100 Gbps within a region) reduces round trips but increases reducer memory pressure.

Shuffle spill happens when the in-memory buffer on the reducer side exceeds the available execution memory. The spill-to-disk path uses Kryo or Java serialization depending on configuration. Spill is a fallback, not a design target: if a job spills heavily in every stage, the partition count is too low. The rule of thumb is 128–256 MB per partition; for a 1 TB dataset that implies 4,000–8,000 partitions.

The Catalyst Query Optimizer

Catalyst is the optimizer embedded in Spark SQL and DataFrame APIs. It operates on a logical plan tree and transforms it through four phases: analysis, logical optimization, physical planning, and code generation.

Analysis resolves attribute references and types against the catalog. An unresolved plan like SELECT id, name FROM users WHERE age > 30 becomes a resolved plan where users is a specific table, id has type LongType, and age has type IntegerType.

Logical optimization applies rule-based rewrites. These rules fire in fixed-point iteration until no rule produces a change. Examples include:

• Predicate pushdown: move filters as close to the data source as possible, so Parquet column readers can skip entire row groups.
• Constant folding: replace WHERE 1 = 1 with true at plan time, not at row evaluation time.
• Column pruning: eliminate columns from scans that are not referenced downstream.
• Null propagation: simplify expressions like x + null to null at plan time.

Physical planning converts the optimized logical plan into one or more physical plans, cost-estimates them using table statistics, and selects the best. This is where join strategy selection happens: broadcast hash join (small table fits in memory), sort-merge join (both sides sorted on join key), shuffle hash join (one side much smaller), or Cartesian product (no join key).

// TypeScript pseudocode: Catalyst rule-based optimization
interface LogicalPlan {
  type: string;
  children: LogicalPlan[];
}

type OptimizerRule = (plan: LogicalPlan) => LogicalPlan;

function applyRulesFixedPoint(
  plan: LogicalPlan,
  rules: OptimizerRule[]
): LogicalPlan {
  let current = plan;
  let changed = true;
  while (changed) {
    changed = false;
    for (const rule of rules) {
      const next = rule(current);
      if (!planEquals(next, current)) {
        current = next;
        changed = true;
      }
    }
  }
  return current;
}

const predicatePushdown: OptimizerRule = (plan) => {
  if (plan.type === "Filter" && plan.children[0].type === "Project") {
    // push filter below projection if it only references projected columns
    return pushFilterBelowProject(plan);
  }
  return plan;
};

Code generation (via Janino or the newer Wholestage CodeGen in Tungsten) compiles the physical plan into JVM bytecode. Instead of a virtual dispatch per row per operator, the generated code processes a batch of rows in a tight inner loop. For CPU-bound workloads on numeric columns, this approaches C-like throughput from the JVM.

Tungsten: Memory and CPU Efficiency Below the API

Tungsten was introduced in Spark 1.5 to address two JVM limitations: GC overhead from millions of short-lived objects, and the CPU cost of generic expression evaluation.

Off-heap binary storage is the core idea. Instead of storing rows as Java objects (which carry object headers, type references, and pointer indirection), Tungsten stores rows as raw binary in a contiguous memory region managed outside the JVM heap. A row is a compact bitset of null flags followed by fixed-width fields (primitives) and offsets into a variable-length section for strings and arrays.

The benefit is cache friendliness and zero GC pressure for shuffle and join buffers. A sort on a 1 billion-row dataset that would create 1 billion Java objects (and trigger major GC every few seconds) instead operates on a single large off-heap buffer.

Memory in a Spark executor is divided into three regions:

1. Reserved memory (300 MB fixed): reserved for internal Spark structures.
2. User memory (governed by 1 - spark.memory.fraction, default 40%): available for user data structures, UDF state, and anything outside Spark’s memory manager.
3. Unified memory (governed by spark.memory.fraction, default 0.6): shared between execution memory (shuffle buffers, sort buffers, join hash tables) and storage memory (cached RDD/DataFrame blocks). The boundary between execution and storage within unified memory is dynamic: execution can borrow from storage (evicting cached blocks if necessary), but storage cannot evict execution memory.

Executor Heap (e.g., 8 GB)
├── Reserved: 300 MB
├── User memory: 3.08 GB  (0.4 × (8GB - 300MB))
└── Unified memory: 4.62 GB  (0.6 × (8GB - 300MB))
    ├── Execution: can expand up to 4.62 GB (evicting storage)
    └── Storage: can expand up to 4.62 GB (but yields to execution)

The implication: on jobs with no caching, spark.memory.storageFraction (the minimum guaranteed to storage within unified memory, default 0.5) is wasted. Set it lower (0.1–0.2) on pure-processing jobs to give more headroom to execution, reducing shuffle spills.

Adaptive Query Execution

Adaptive Query Execution (AQE), enabled by default in Spark 3.0+, re-optimizes the query plan at shuffle boundaries using runtime statistics.

At each stage boundary, AQE collects the actual sizes and row counts written by map tasks. This information was not available during initial planning (the optimizer relied on table statistics that may be stale or absent). AQE uses it to make three categories of decisions:

Coalescing post-shuffle partitions: if the target partition count was 200 but most partitions contain only a few KB, AQE merges adjacent small partitions into larger ones. This prevents the overhead of thousands of near-empty tasks in the downstream stage. The target size is controlled by spark.sql.adaptive.advisoryPartitionSizeInBytes (default 64 MB).

Converting sort-merge joins to broadcast joins: if one side of a join is revealed to be small at runtime (below spark.sql.autoBroadcastJoinThreshold, default 10 MB), AQE replaces the planned sort-merge join with a broadcast hash join, eliminating a full shuffle on the larger side.

Skew join optimization: if one partition in a join is significantly larger than the median (controlled by spark.sql.adaptive.skewJoin.skewedPartitionFactor, default 5x), AQE splits the skewed partition and replicates the matching partition from the other side. This turns a single straggler task that runs 50x longer than its siblings into multiple balanced tasks.

// TypeScript pseudocode: AQE runtime feedback loop
interface StageStatistics {
  stageId: number;
  partitionSizes: number[];  // bytes per output partition
  rowCounts: number[];       // rows per output partition
}

class AdaptiveQueryExecution {
  onStageComplete(stats: StageStatistics, plan: PhysicalPlan): PhysicalPlan {
    let optimized = plan;
    optimized = this.coalesceSmallPartitions(optimized, stats);
    optimized = this.upgradeTobroadcastJoin(optimized, stats);
    optimized = this.handleSkewedJoin(optimized, stats);
    return optimized;
  }

  private coalesceSmallPartitions(
    plan: PhysicalPlan,
    stats: StageStatistics
  ): PhysicalPlan {
    const advisorySize = 64 * 1024 * 1024; // 64 MB
    const groups = groupPartitionsToTargetSize(stats.partitionSizes, advisorySize);
    return rewritePlanWithCoalescedPartitions(plan, groups);
  }

  private handleSkewedJoin(
    plan: PhysicalPlan,
    stats: StageStatistics
  ): PhysicalPlan {
    const median = medianOf(stats.partitionSizes);
    const skewFactor = 5;
    const skewedPartitions = stats.partitionSizes
      .map((size, idx) => ({ size, idx }))
      .filter(({ size }) => size > median * skewFactor);
    if (skewedPartitions.length > 0) {
      return splitSkewedPartitions(plan, skewedPartitions);
    }
    return plan;
  }
}

AQE changes the failure model slightly: a stage that was expected to run in parallel may now be re-planned at its boundary, adding a brief planning pause. For interactive queries this is invisible; for batch jobs with tens of stages it is worth monitoring in the Spark UI’s stage timeline view.

Production Considerations

Partition sizing. The most common production problem is wrong partition counts. Too few partitions means large tasks that spill and cannot parallelize. Too many means excessive task scheduling overhead and tiny shuffle files. Target 128–256 MB per partition for raw data reads and 64–128 MB per partition after aggregations. Use df.repartition(n) or df.coalesce(n) deliberately rather than relying on defaults.

Serialization format. Kryo serialization (spark.serializer=org.apache.spark.serializer.KryoSerializer) is 10–40% faster than Java serialization for most workloads and produces significantly smaller shuffle data. Register classes explicitly with spark.kryo.classesToRegister to avoid per-object class metadata in the serialized form. For DataFrames and Datasets, Tungsten’s binary format already bypasses Java serialization, but RDD operations and UDFs still use the configured serializer.

Join strategy selection. Broadcast joins are the cheapest join type when one side fits in memory. The threshold is spark.sql.autoBroadcastJoinThreshold (default 10 MB, try 100–500 MB on clusters with sufficient executor memory). AQE handles cases where size is unknown at planning time. For very large join keys with skew, broadcast the small side rather than fighting sort-merge join stragglers.

Memory fractions. The defaults work for mixed workloads. Adjust spark.memory.fraction (default 0.6) only when you have evidence of spills (look for “spilled” in stage metrics) or of OOM errors attributable to execution memory. Increasing to 0.7–0.75 on pure-processing jobs with no caching is usually safe. Do not set it above 0.8 unless user memory usage is provably zero.

Off-heap memory. Enable spark.memory.offHeap.enabled=true and set spark.memory.offHeap.size for workloads with large shuffle or sort operations. Off-heap storage removes these buffers from GC accounting, reducing pause frequency. The tradeoff is that off-heap OOM errors are harder to diagnose than heap OOM errors.

Speculation and stragglers. spark.speculation=true re-launches slow tasks on other executors. It helps with hardware-related stragglers but creates duplicate work and can exacerbate shuffle congestion. Enable it only after verifying that stragglers are not caused by data skew, because speculation does not fix skew and the redundant tasks will also be slow.

Dynamic allocation. spark.dynamicAllocation.enabled=true releases executors after idle periods and requests new ones when the task queue grows. Combined with the external shuffle service, this significantly reduces cluster cost for batch jobs with variable parallelism across stages. Set spark.dynamicAllocation.minExecutors to avoid cold-start latency on small-stage bursts.

Tradeoffs: Spark vs. Alternatives

Dimension	Apache Spark	Apache Flink	Trino	Dask	Ray
Processing model	Micro-batch and batch; structured streaming for near-real-time	True record-at-a-time streaming; batch via DataStream API	Interactive SQL query engine; no native streaming	Python-native task graph; batch and streaming	General distributed compute; actor model for state
State management	In-memory caching; external state via Delta/Iceberg	Native keyed state with configurable backends; exactly-once	Stateless per query; no built-in state across queries	Limited; Dask Arrays/DataFrames carry no cross-partition state	Actors hold state; no automatic checkpointing
Latency profile	Seconds to minutes; structured streaming near sub-second	Sub-second to milliseconds	Sub-second for cached queries; minutes for large scans	Seconds to minutes	Sub-second for small graphs; minutes for large ML jobs
Ecosystem	Broadest: Delta Lake, MLlib, GraphX, Spark SQL	Kafka, Flink CDC, Hudi, PyFlink; ML ecosystem thinner	Hive, Iceberg, Delta, Hudi connectors; read-heavy	NumPy, Pandas, scikit-learn; Python-first	PyTorch, TensorFlow; ML training focus
Fault tolerance	Re-run from lineage or checkpoint; shuffle files on local disk	Chandy-Lamport checkpoints; fast incremental recovery	None (queries re-run); compute tier is ephemeral	Re-run from lineage; no checkpointing	Lineage-based; actors can checkpoint manually
Scaling model	Static or dynamic allocation; coarse-grained executor model	Fine-grained task slots; scales to per-key parallelism	Query-level parallelism; worker pool shared across queries	Cluster per session or per notebook; less isolation	Cluster shared; job-level resource groups
SQL maturity	High; ANSI SQL + extensions; Catalyst optimizer	Flink SQL growing; table API complete	Highest; full ANSI SQL; optimized for analytics	pandas-like API; no SQL engine	None native; use with Ray Data + DuckDB
Sweet spot	Large-scale batch ETL, ML pipelines, data lake transforms	Event-driven streaming, fraud detection, real-time aggregations	Ad-hoc interactive analytics on data lake at petabyte scale	Python-native data science teams, moderate-scale ML	Distributed ML training, reinforcement learning, online inference

Closing

Spark’s architecture is a consequence of one foundational bet: lazy evaluation over immutable distributed datasets, with shuffles as explicit stage boundaries. The DAG scheduler, Catalyst optimizer, Tungsten memory model, and AQE runtime feedback all exist to make that bet pay off at scale. When a job misbehaves in production, the answer is almost always in one of these layers: the wrong number of partitions crossing a shuffle boundary, a join strategy chosen without runtime size information, or execution memory starved by a storage cache that was never needed. Knowing the mechanism makes the diagnosis systematic rather than exploratory.