Apache Spark
Design Spark workloads with partition sizing, shuffle and skew control, executor memory, retries, lineage recovery, and production operations.
What is Apache Spark?
Apache Spark is a distributed computation engine that turns a data job into tasks and runs those tasks across a cluster. It is commonly used for batch ETL, interactive SQL, feature preparation, and stateful stream processing when one machine cannot finish the work inside the required time or memory budget.
Spark matters because it gives several data workloads one execution model:
- a driver builds and coordinates the plan;
- executors run parallel tasks and hold cached or shuffle data;
- a cluster manager supplies processes and resources;
- storage systems remain the durable source and destination.
The core invariant is simple: Spark can parallelize independent partitions, but a dependency that needs records from other partitions creates data movement and a new scheduling boundary. More executors do not automatically fix skew, excess shuffling, tiny files, an overloaded driver, or an unsafe streaming sink.
This lesson moves from that mental model to stage planning, partition sizing, executor failure, Structured Streaming, and production operations. Review distributed systems first if partitions, retries, and failure domains are unfamiliar.
Follow one job from code to tasks
A Spark application is one driver plus a set of executor processes. The application can submit many jobs over its lifetime.
The Spark control and data path
The driver coordinates work; executors exchange and transform partition data.
Logical intent
Application code
DataFrame, SQL, or Dataset operations describe filters, joins, aggregations, and output without manually assigning records to machines.
Plan and coordinate
Driver
Builds the logical plan, asks Spark to optimize it, creates stages and tasks, and tracks their completion. Driver loss normally ends the application unless the platform restarts it.
Allocate processes
Cluster manager
Kubernetes, YARN, or a standalone cluster grants resources for the driver and executors. It does not choose Spark's join or partition strategy.
Run tasks
Executors
Each executor runs several tasks concurrently, keeps cached blocks, writes shuffle output, and reports results and metrics to the driver.
Four terms keep the execution hierarchy clear:
- Application: one driver and its executors.
- Job: work triggered by an action such as writing data or counting rows.
- Stage: a group of tasks that can run without crossing a shuffle boundary.
- Task: one stage's work for one partition.
An action does not run "on the cluster" as one indivisible operation. It becomes many tasks whose duration, memory demand, and data placement determine the job's real behavior.
Read the DAG before tuning the cluster
Spark transformations are lazy: they describe a logical plan and do not start execution until an action requires a result. Spark then optimizes the plan and builds a directed acyclic graph, or DAG, of dependencies.
One parent partition
Narrow dependency
Each output partition reads from one parent partition. Filters, projections, and many row-level transformations can be pipelined in one stage without redistributing records across executors.
Many parent partitions
Wide dependency
An output partition needs records from several parent partitions. Grouping, sorting, repartitioning, and many joins write and fetch shuffle blocks, so they create a stage boundary.
1 Lazy transformations
Build the logical plan
Record the requested filters, projections, joins, and aggregations without materializing each intermediate result.
2 Catalyst and strategy selection
Optimize the physical plan
Push filters, prune columns, choose operators, and decide whether a join can avoid a full shuffle. Inspect the actual plan instead of assuming the choice.
3 Dependency boundary
Split stages at shuffles
Pipeline narrow operations together. Materialize exchange output where records must be regrouped by key, range, or partition.
4 Parallel execution
Run one task per partition
Schedule tasks into available executor slots, retry failed attempts, and start downstream stages when their required shuffle outputs exist.
A stage with 2,000 partitions exposes at most 2,000 concurrent tasks, even if the cluster has more free slots. A stage with 20 partitions cannot use 200 cores at once. Partition count therefore controls both parallelism and the size of each unit of work.
Lab 1: turn a logical operation into a stage plan
Choose a workload and execution strategy, then change partition count and hot key concentration. The lab shows four consequences together:
- shuffle bytes crossing the network;
- average and hottest partition size;
- task waves across the available slots;
- whether the chosen plan is valid for the data shape.
Use the failure states deliberately. Try broadcasting a dimension that exceeds the modeled safety limit, then compare a plain hash aggregation with a salted hot key.
Loading the Spark stage model
The lesson is loading its co-located partition and shuffle contract.
Loading stage model...
The goal is not the smallest partition or the fewest shuffle bytes in isolation. Choose enough partitions to use the cluster while keeping scheduler overhead, per-task memory, skew, and output-file count inside measured limits.
Choose the highest-level API that preserves the plan
DataFrames, SQL, Datasets, and RDDs all execute on Spark, but they expose different information to the optimizer.
Default for structured data
DataFrame or SQL
Use named columns and relational operators so Spark can prune columns, push filters, reorder work, and select physical operators. Validate the optimized and physical plans for important jobs.
JVM type boundary
Typed Dataset
Use when Scala or Java code benefits from compile-time domain types and the encoder cost is acceptable. Some typed operations can reduce optimizer visibility compared with relational expressions.
Low-level control
RDD
Use for data or algorithms that do not fit structured operators. RDD code owns more serialization, partitioning, and aggregation decisions and gives Catalyst less structure to optimize.
Preserve useful structure before adding resources:
- project only required columns before a wide exchange;
- filter early when semantics permit it;
- pre-aggregate by key before a large shuffle;
- broadcast only a bounded, measured side of a join;
- align partitioning when several repeated operations use the same key;
- avoid Python row UDFs when a built-in expression or vectorized operation can express the same logic;
- compact small source and output files outside latency-sensitive jobs.
cache() is not a universal accelerator. Persist an intermediate result only when it is reused and recomputing it costs more than storing, serializing, and evicting it. Unpersist it after the final consumer.
Estimate partitions from bytes, slots, and skew
Start with a target task size, then validate it against real stage metrics. For an illustrative 600 GiB shuffle:
600 GiB
Shuffle input
Bytes after projection and pre-aggregation
256 MiB
Initial target
One measured task-size hypothesis
2,400
Starting partitions
600 GiB / 256 MiB
200 slots
Peak concurrency
About 12 waves before skew
The calculation is a starting point, not a Spark constant. Adjust it when:
- compression makes input bytes differ from in-memory row size;
- one key concentrates far more than the average partition;
- per-task state, sorting, or hash tables spill despite moderate input bytes;
- remote storage or an external service limits useful concurrency;
- the sink cannot tolerate thousands of small output files;
- adaptive query execution coalesces or splits shuffle partitions at runtime.
Skew requires a key-level response. Adding partitions helps only if the partitioner can spread the hot records. Salting, skew-aware joins, pre- aggregation, or isolating a dominant key changes that distribution; adding idle executors does not.
Size executors around the task, not a round number
An executor has heap memory, non-heap overhead, CPU cores, and a local failure boundary. More cores per executor create more concurrent tasks sharing one heap, garbage collector, network connection pool, and process fate.
Reserve memory for distinct consumers:
- execution memory for joins, sorts, aggregations, and shuffles;
- storage memory for cached blocks and broadcasts;
- user memory for objects not managed by Spark's execution pool;
- process overhead for native memory, Python workers, and container costs;
- headroom for skewed tasks and transient serialization buffers.
Fewer large executors reduce process overhead but increase the number of task slots and cached blocks lost with one process. Many small executors reduce blast radius but add scheduling, connection, and per-process overhead.
Lab 2: expose executor failure blast radius
Select an executor shape and a failure event. Change executor count and task duration to see how process size affects:
- usable task memory and spill pressure;
- total parallel slots and task waves;
- tasks invalidated by executor loss;
- estimated recomputation and recovery time.
Loading the executor failure model
The lesson is loading its co-located executor and stage contract.
Loading executor model...
Choose a shape from measured task memory and garbage-collection behavior, then reserve enough surviving capacity for the failure mode the service-level objective assumes.
Make retries safe at every boundary
Spark can recompute a lost partition from lineage and retry failed task attempts. That protects deterministic computation, not every external effect.
One partition attempt
Task retry
Safe when the transformation is deterministic and the attempt does not create an untracked external side effect. A retried task may run more than once.
Lost executor output
Shuffle recovery
If required shuffle blocks disappear with an executor, Spark may rerun upstream map tasks before downstream work can continue. External shuffle storage or decommissioning can change the blast radius.
Application coordination
Driver recovery
Executor retries do not preserve driver-local state. Restart behavior depends on the cluster platform, durable checkpoints, source offsets, and idempotent sink design.
Speculative execution can launch a duplicate attempt for a slow task. It helps with an unhealthy worker, but not when every copy receives the same hot key or waits on the same overloaded dependency.
Checkpointing truncates lineage by writing a durable recovery point. Use it for long or stateful dependency chains where recomputing from the source would violate the recovery objective. Do not confuse a cached block with a durable checkpoint.
Build a partition-aware batch pipeline
The example keeps projection and filtering before the join, validates the dimension's bounded size before allowing a broadcast hint, and repartitions the final write by a business key. The numbers are explicit hypotheses to measure, not universal defaults.
Review the plan and stage metrics after representative runs:
- Confirm source filters and column pruning appear in the physical plan.
- Check whether the intended broadcast or shuffle join was actually selected.
- Compare median and maximum task input, duration, spill, and shuffle fetch.
- Verify output file sizes and partition distribution at the sink.
- Repeat with production-shaped skew and one executor unavailable.
Treat Structured Streaming as repeated, stateful computation
Structured Streaming expresses a continuously updated result as incremental table processing. A trigger selects available input, Spark plans tasks for that trigger, and a checkpoint records progress and state metadata needed for recovery.
1 Source offsets
Read bounded progress
Discover the next input range from a replayable source. Persist enough source progress to reconstruct what a restarted query should process.
2 Event time
Transform and update state
Apply parsing, deduplication, windows, joins, and aggregations. Watermarks bound how long old event-time state must remain eligible for late data.
3 Retry-safe output
Commit the sink
Write a deterministic batch or transaction identifier so replay does not duplicate a business effect. Sink guarantees matter as much as Spark retries.
4 Recovery evidence
Checkpoint progress
Record committed offsets and state-store metadata in durable, isolated storage. Monitor checkpoint latency and verify restart from a production snapshot.
Watermarks are state-retention and completeness decisions, not a promise that all records arrive on time. A shorter delay reduces state but can finalize windows before late events; a longer delay preserves more late data but grows state and recovery work.
Operate Spark from stage evidence
Cluster CPU alone cannot explain a slow job. Preserve application, job, stage, task, executor, and source/sink dimensions in telemetry.
Watch the execution path
- scheduling delay and active versus pending tasks;
- stage duration percentiles and critical-path stages;
- median, p95, and maximum task duration;
- input, output, shuffle read, and shuffle write bytes;
- local and remote shuffle fetch wait;
- memory and disk spill per task;
- executor lost, task retry, fetch failure, and speculation counts.
Watch the data shape
- partition row and byte distribution;
- hottest keys and skew ratio;
- source file count and average file size;
- output file count, size distribution, and failed commits;
- cache hit, eviction, and recomputation behavior;
- streaming input rate, processing rate, state rows, watermark, and trigger lag.
Rehearse recovery
- Remove one executor during the largest shuffle.
- Restart the driver from the production checkpoint and deployment contract.
- Throttle the source or sink and verify backpressure and alert timing.
- Replay a streaming batch and prove the sink does not duplicate effects.
- Inject one hot key and confirm the maximum task, not the average, triggers the capacity response.
- Restore checkpoint and source credentials in a clean environment.
Keep the Spark event log long enough to reconstruct an incident. A completed application disappears from the live UI, but its stage and task evidence is still needed for regression analysis.
Choose Spark when distributed execution is the real need
Distributed transformation
Spark is a strong fit
- Data exceeds one machine's practical memory, CPU, or completion window.
- The workload benefits from SQL, batch, and streaming under one execution model.
- Transformations can be expressed as deterministic partition work.
- The team can inspect plans, operate shuffles, and test recovery.
Different constraint
Use a simpler or different engine
- One machine or an embedded analytical database finishes reliably and more cheaply.
- Millisecond request serving is the primary workload.
- The computation needs tightly coupled low-latency updates between every worker.
- External side effects cannot be made idempotent under task retry.
Production review
Before launch, require evidence for:
- Plan: important joins, exchanges, partition counts, and adaptive behavior are visible in representative physical plans.
- Data: skew, row width, file count, and growth assumptions use measured production-shaped samples.
- Resources: executor memory, cores, overhead, and surviving capacity are tied to task metrics and a named failure.
- Correctness: retries, speculative attempts, and streaming replay cannot duplicate or lose a business effect.
- Recovery: executor, shuffle, driver, source, sink, and checkpoint failures have rehearsed responses.
- Operations: event logs, stage metrics, data-quality checks, and ownership remain available after the application exits.