Engineering Atlas › System Design Interviews

Apache Spark for system design interviews

13 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

How it works

Driver, executors and the cluster manager

The driver runs your main, holds the SparkSession, and plans and schedules work. A cluster manager (YARN, Kubernetes or Spark standalone) launches executors, JVMs with a set number of cores and memory. Each core runs one task at a time, so 100 executors with 4 cores each run 400 tasks concurrently. With Spark Connect (since 3.4) the application is a thin client that sends unresolved logical plans to the driver over gRPC and gets Arrow batches back, so a misbehaving notebook no longer shares the driver's JVM; it gives up the RDD and SparkContext APIs (Spark Connect overview).

From DataFrame to physical plan

DataFrame and SQL code is parsed into a logical plan, which the Catalyst optimizer rewrites: predicate pushdown into Parquet scans, column pruning, constant folding, join reordering. The planner picks physical operators such as sort-merge join or broadcast hash join. Whole-stage code generation then fuses chains of operators into a single generated Java method per stage, avoiding virtual calls per row. Adaptive Query Execution (on by default since 3.2) re-plans at shuffle boundaries using the actual sizes of map outputs (SQL performance tuning). Since 4.0, Spark SQL also runs in ANSI mode by default (spark.sql.ansi.enabled=true): an overflow or invalid cast raises an error at runtime instead of quietly returning null (Spark 4.0 release notes, configuration).

Narrow and wide dependencies

A narrow dependency means each output partition depends on a bounded set of input partitions on the same task: map, filter, union, reading a file. These pipeline inside one task with no data movement. A wide (shuffle) dependency means an output partition needs data from many input partitions: groupByKey, reduceByKey, join on differently partitioned inputs, repartition, distinct. The DAGScheduler walks the lineage backwards from the action and starts a new stage at each shuffle dependency.

Caching and storage

cache() or persist() keeps computed partitions in executor memory or on local disk via the BlockManager, so later jobs skip recomputing them. Output normally goes to object storage in columnar formats (Parquet, ORC) or table formats (Delta Lake, Iceberg, Hudi) that add transactions on top.

Down the stack

Layer 1The contract

Spark promises that an action produces the same result as running the transformation sequentially, as long as your functions are deterministic, regardless of executor failures: lost tasks are retried (spark.task.maxFailures, default 4 consecutive failures of one task) (configuration) and lost shuffle output is recomputed by rerunning parent stages. It does not promise exactly-once side effects from inside tasks, because retried or speculative tasks run your code more than once; writes are made safe by output committers and table formats that commit atomically at the end. Latency is batch-shaped: a job with several shuffles over terabytes takes minutes. Structured Streaming micro-batches can get as low as about 100 ms end to end with exactly-once guarantees, and usually run at seconds (Structured Streaming guide). Spark 4.1 added a real-time mode for continuous, sub-second processing, initially for stateless Scala queries (Spark 4.1 release notes).

Layer 2The mechanism

Stage construction. For the job's final RDD the DAGScheduler creates a ResultStage and walks dependencies: narrow ones join the current stage, and each ShuffleDependency creates or reuses (by shuffle ID) a ShuffleMapStage. A stage is submitted once its parents finish, as a TaskSet with one task per partition. The TaskScheduler prefers executors holding the input (PROCESS_LOCAL, NODE_LOCAL, then ANY), waiting up to spark.locality.wait (3 seconds per locality level) for a better slot.

Sort-based shuffle. Each map task partitions records by the reducer's partitioner (hash of key modulo N), buffering in memory and spilling sorted runs to disk when full. It then merges spills into one data file plus an index of byte offsets per reduce partition: shuffle_<shuffleId>_<mapId>_0.data and .index. The driver's MapOutputTracker records each map output's location and block sizes. Reducers fetch blocks over Netty from remote BlockManagers or the external shuffle service, capped by spark.reducer.maxSizeInFlight (48 MiB). A failed fetch raises FetchFailed, and the parent map stage is resubmitted for the missing outputs. That partial rerun is only correct if the map stage reproduces the same output; Spark 4.1 added a checksum-based full retry of a shuffle stage to catch the cases where it does not (Spark 4.1 release notes).

AQE. After a map stage, AQE reads real partition sizes, coalesces small shuffle partitions (so the default spark.sql.shuffle.partitions=200 matters less), splits skewed join partitions, and switches to a broadcast hash join when a side is under spark.sql.autoBroadcastJoinThreshold (10 MB) (SQL performance tuning).

Memory. spark.memory.fraction (0.6) of heap minus 300 MB is shared by execution and storage; execution can evict cached blocks down to spark.memory.storageFraction (0.5) (Tuning Spark). Tungsten keeps rows in compact binary form, on or off heap.

Layer 3The machine

A Parquet scan reads columns in large sequential ranges, decompresses (Snappy or Zstd) and decodes into vectorised batches; one core can scan several hundred MB/s of compressed data for simple filters (hedge: depends heavily on schema and codec). Object storage adds 20 to 100 ms of first-byte latency per request, which is why files under about 32 MB hurt and 128 MB to 1 GB files work well; spark.sql.files.maxPartitionBytes (128 MB) sets the split size.

With M map tasks and R reduce partitions a shuffle has M x R blocks: 2,000 x 2,000 is 4 million small random reads. On HDDs at 100 to 200 IOPS that is catastrophic; on NVMe it is tolerable. A 2 TB shuffle on 100 nodes with 25 Gbit/s NICs moves 20 GB per node in about 7 seconds of raw transfer, so serialisation, compression and spills usually dominate.

Very large heaps (over about 64 GB) risk multi-second GC pauses, so executors are commonly 4 to 5 cores and 16 to 40 GB with G1 GC.

Why it's built this way

  1. MapReduce wrote every intermediate step to replicated HDFS, which was slow for multi-step jobs, so Spark keeps intermediate data in memory and pipelines narrow transformations.
  2. Keeping data only in memory means it can be lost, so Spark records lineage and recomputes lost partitions instead of paying for replication on every step (RDD paper).
  3. Some operations need all values for a key in one place, which requires moving data across the network; shuffle boundaries are therefore natural checkpoints, and map outputs are written to local disk so a failed reducer does not rerun the whole job.
  4. Static plans guess wrong about data sizes, so AQE re-plans at each shuffle boundary, where exact sizes are finally known.
  5. JVM objects waste memory and stress the GC, so Tungsten stores binary rows and generates code that operates on them directly.

Diagram

flowchart LR
    subgraph S0["Stage 0 (ShuffleMapStage)"]
      R["read Parquet"] --> F["filter"] --> M["map to (key, 1)"]
    end
    M --> SW["shuffle write: .data + .index per map task"]
    SW --> MOT["MapOutputTracker (driver)"]
    subgraph S1["Stage 1 (ResultStage)"]
      FE["shuffle fetch blocks"] --> AGG["reduceByKey"] --> W["write output"]
    end
    MOT --> FE
    SW --> FE

Worked example

A nightly job joins 3 TB of clickstream events (Parquet, 30 days) with a 40 GB user table and a 200 MB country table, then aggregates sessions per user per day.

Scanning 3 TB at 128 MB per split is about 24,000 tasks; on 250 cores at about 10 seconds per task, that is 96 waves, roughly 16 minutes. Pushing the date filter into Parquet and reading 8 of 60 columns cuts bytes read to about a quarter.

The country table is broadcast (200 MB to 50 executors, 10 GB of network), so its join needs no shuffle. The user join is a sort-merge join with both sides shuffled by user_id; the events side shuffles about 600 GB. With the default 200 partitions that is 3 GB per reduce task, which spills heavily, so the job starts at 3,000 partitions (200 MB each) and lets AQE coalesce. A bot user with 2 percent of events creates a 12 GB partition; AQE skew handling splits it, or bots are filtered first. The final groupBy(user_id, date) reuses the user_id partitioning and does not shuffle again.

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
SparkMature, SQL plus code, scales to petabytes, rich ecosystemShuffle and JVM tuning, seconds of overhead per jobLarge batch ETL, joins, ML feature pipelines
FlinkTrue streaming, event time, large stateSmaller batch ecosystemLow-latency stateful streaming
Warehouse SQL (BigQuery, Snowflake, Redshift)No cluster tuning, fast interactive SQLCost at very large scale, less control over custom codeSQL analytics and BI
Trino or PrestoFast interactive federated queriesNot built for long fault-tolerant ETLAd hoc queries over a lake
Single node (DuckDB, Polars)No distribution overhead, very fast up to hundreds of GBOne machine's limitsData fits on one large box

Pitfalls in production

Leadership lens

Check yourself

  1. Why does reduceByKey usually shuffle less data than groupByKey followed by a sum?

    Show answer

    reduceByKey combines values per key on the map side before the shuffle, so each map task sends one partial sum per key instead of every value.

  2. A reduce task fails with FetchFailed. What does Spark do?

    Show answer

    It marks that location's map outputs lost, reruns the parent map stage for just those outputs, then retries the reduce stage.

  3. Why do two joins on the same key not always need two shuffles?

    Show answer

    If both inputs of the second join are already hash-partitioned on that key with the same partition count, the dependency is narrow and Spark reuses the existing partitioning.

  4. Why is a single giant executor with 64 cores and 512 GB heap usually worse than many smaller ones?

    Show answer

    Large heaps cause long GC pauses, and one failure loses much more work and cached data.

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →