Spark — RDD, DataFrame, Jobs/Stages/Shuffles

Why this session exists

Spark tuning begins with an execution model, not “add executors.” This lesson traces DataFrame expressions through logical/physical plans, jobs, stages, tasks and exchanges, then reasons about skew, broadcast, caching and small files. It targets Spark 3.5.7. The bounded Python model is executed; the PySpark reference is source-reviewed without starting Spark or claiming cluster performance.

Outcomes: explain laziness and actions; distinguish task count from concurrency; trace shuffle dependencies; diagnose skew using bytes/records as well as duration; choose safe join strategies; and demonstrate correct salted aggregation and joins against independent oracles.

Prerequisites

Joins supplies bag semantics and output cardinality; data modelling supplies grain; batch versus streaming separates input semantics from execution technique. Spark SQL resembles SQL but has engine-specific NULL, typing and optimization behavior to verify.

A distributed cooking brigade

Imagine chefs processing ingredient crates. One giant crate delays completion even if other chefs are idle: skew. Moving ingredients among chefs resembles exchange. Writing a menu before cooking resembles lazy transformations. The analogy does not establish that 40% of time is movement or that 90% of performance is three knobs.

RDD transformations construct dependencies; actions such as count and collect trigger computation for a result. DataFrames add schema and relational expressions that Catalyst can analyze, simplify, reorder where legal, prune and physically plan. RDD closures are comparatively opaque to Catalyst, but RDDs remain useful for custom partition-level algorithms. A Python UDF in a DataFrame can also hide logic from SQL optimization.

explain can create/inspect a physical plan before an action. Some API construction, schema inference or metadata reads can itself do work; laziness is not “no work of any kind before count.” A DataFrame describes a logical dataset, not necessarily already materialized in executor memory. Cache/persist is lazy until materialized; eviction can cause recomputation. Use it when reuse saves enough work to justify memory and spill, then unpersist deliberately.

Historical context and current defaults

RDD lineage and partitioned execution preceded the structured SQL/DataFrame layer. Spark 3.0 expanded Adaptive Query Execution, and Spark's 3.5.7 guide states AQE has been enabled by default since 3.2.0. Spark Connect was introduced in 3.4, not 3.5. These are API/execution milestones, not proof of universal speedups or that Spark “killed Hadoop.” Defaults and supported operators must be tied to the selected release.

From code to cluster

An action can involve multiple jobs, especially with subqueries, broadcasts and adaptive execution. A stage has tasks for the partitions it processes; retries/speculation add attempts. Eight partitions on two available slots do not mean eight simultaneously running tasks. The DAG can branch and reuse completed stages, so stage count is not universally shuffle count plus one. That rule only describes a simple unreused linear chain.

Anatomy of a shuffle and the dependency derivation

Spark 3.5.7 defines a narrow dependency from the child side: each child partition depends on a small number of parent partitions, allowing pipelined execution. Map/filter are simple examples, and coalescing can combine several parents without a full shuffle [tagged Dependency.scala]. Compatible operations can be pipelined in a task. Narrow does not guarantee all input is already local or that no network I/O occurs—storage reads can be remote.

A shuffle redistributes records by a partitioning rule so downstream tasks receive their required groups. Upstream map outputs must be available for dependent downstream work. This creates scheduler dependencies and materialized intermediate data, but it does not mean the entire cluster stops at one universal barrier. Independent branches can progress, and SQL broadcast/AQE exchanges complicate the simple linear picture [Spark RDD].

With N maps and M reducers there can be N×M logical map-to-reducer blocks, not necessarily N×M separate physical files or network requests. Spark 3.5.7's index shuffle resolver describes one data file per map output with reducer blocks and a separate index file; implementation paths and shuffle services can vary. Network volume is measured in bytes, with locality, compression, combining and fetch batching affecting actual transfer. The original diagram's “M files per map” was not an implementation-independent fact.

Local partial aggregation can shrink a million repeated keys into far fewer partial counts before shuffle. Therefore a hot-key count is not automatically the same performance problem as a huge duplicate-producing join. Filter/project before exchange when semantics allow; moving a filter across an outer join can change the answer.

Join strategies and their boundaries

The 3.5.7 automatic broadcast threshold is 10 MiB (10×1024×1024 bytes); eligible estimated size is a planning input, not a runtime memory guarantee. A broadcast hint can prioritize broadcast even above the automatic threshold but does not support every join type. Removing a hint may still yield broadcast. An oversized hinted table can OOM; crossing an automatic threshold may instead change strategy. AQE can use runtime statistics to coalesce partitions, change join strategy and optimize supported skewed joins; it does not fix every skewed aggregation or provide a guaranteed percentage speedup [Spark SQL tuning].

Read the Spark UI as evidence

1. Jobs: identify the action, duration and stage DAG; distinguish waiting from execution.
2. Stages: inspect input/shuffle bytes, records, spill, scheduler delay and task duration distributions.
3. Tasks: compare max/median/percentiles and per-task input sizes. A 5s max versus 0.2s median gives 25× duration ratio, a clue—not proof of data skew. Slow hosts, GC, disk and retries can also create tails.
4. SQL/DataFrame: inspect initial versus final adaptive plans, exchanges and supported join strategy. Broadcast is not inherently “good” and sort-merge “bad.”
5. Executors: compare GC, memory, disk spill, failures and utilization. A fixed GC percentage is not a universal diagnosis.

A single hash key remains together under ordinary hash partitioning even if you double partitions. Salt can split suitable work, but increases phases or replicated data. Many tiny partitions/files can make scheduling and listing overhead dominate; size partitions from data/task duration, memory and slots, not a universal cores×3 or 100–1000 task rule. More workers can help CPU/slot/I/O bottlenecks but cannot split one indivisible task by themselves.

Hands-on: exact salted algorithms and negative controls

This complete CPU program verifies aggregation and bag-join preservation. It does not measure Spark network traffic, AQE behavior or cluster speed. Deterministic salt based on an event identity is stable under replay.

from collections import Counter, defaultdict
from fractions import Fraction
from itertools import product
 
rows = [(i, 7 if i<80 else i%5, i%11) for i in range(100)]
def salted_counts(rows, buckets):
    if buckets<=0: raise ValueError('bucket count')
    partial = Counter((key,eid%buckets) for eid,key,value in rows)
    final = Counter()
    for (key,salt),n in partial.items(): final[key] += n
    return final
expected = Counter(key for _,key,_ in rows)
for buckets in [1,2,4,16]: assert salted_counts(rows,buckets)==expected
assert salted_counts([],4)=={}
# Averages require sum AND count; means of unequal partials are wrong.
groups = [[1],[9,9,9]]
correct_mean = Fraction(sum(map(sum,groups)),sum(map(len,groups)))
wrong_mean = sum(Fraction(sum(g),len(g)) for g in groups)/len(groups)
assert correct_mean==7 and wrong_mean==5
 
def bag_join(events, dims):
    return Counter((eid,key,value,label) for eid,key,value in events for dk,label in dims if key==dk)
def salted_join(events, dims, buckets):
    index = defaultdict(list)
    for key,label in dims:
        for salt in range(buckets): index[(key,salt)].append(label)
    return Counter((eid,key,value,label) for eid,key,value in events for label in index[(key,eid%buckets)])
dims = [(7,'gold'),(7,'alternate'),(0,'free')]
assert salted_join(rows,dims,4)==bag_join(rows,dims)
# Replicating BOTH sides to every salt would multiply every pair by buckets.
assert sum(salted_join(rows,dims,4).values())==expected[7]*2+expected[0]
# Exhaustive small bag cases: duplicate dimension keys are not discarded.
cases=0
for ek in product([0,1],repeat=3):
    for dk in product([0,1],repeat=2):
        e=[(i,k,i+1) for i,k in enumerate(ek)]
        d=[(k,str(i)) for i,k in enumerate(dk)]
        assert salted_join(e,d,3)==bag_join(e,d)
        cases+=1
# The hot key stays together with ordinary hashing; salt only changes the model's groups.
for partitions in [4,8,16]:
    hot_destinations={key%partitions for _,key,_ in rows if key==7}
    assert len(hot_destinations)==1
assert len({eid%4 for eid,key,_ in rows if key==7})==4
assert Fraction(5,1)/Fraction(1,5)==25
print({'counts':sorted(expected.items()),'join_pairs':sum(bag_join(rows,dims).values()),'exhaustive_bag_cases':cases,'weighted_mean':str(correct_mean),'wrong_mean':str(wrong_mean)})

Spark 3.5.7 reference program

The following bounded 1,000-row PySpark program retains the original lazy-plan/action/aggregate/broadcast/skew/salting walkthrough. It is syntax/source-reviewed, not executed. An authorized isolated Spark environment must supply the matching release; no package install, application UI or cluster was started here.

from pyspark.sql import SparkSession, functions as F
spark = (SparkSession.builder.master('local[2]').appName('bounded_execution_lesson')
 .config('spark.ui.enabled','false')
 .config('spark.driver.bindAddress','127.0.0.1')
 .config('spark.driver.host','127.0.0.1')
 .config('spark.sql.adaptive.enabled','true')
 .config('spark.sql.adaptive.skewJoin.enabled','true')
 .config('spark.sql.shuffle.partitions','4')
 .config('spark.sql.autoBroadcastJoinThreshold',10485760).getOrCreate())
try:
    orders = spark.range(1000).selectExpr('id AS order_id',
      'CASE WHEN id < 800 THEN 7 ELSE id % 5 END AS customer_id',
      'id % 100 AS amount')
    orders.explain(mode='formatted')
    assert orders.count()==1000
    counts = orders.groupBy('customer_id').count()
    counts.orderBy('customer_id').show()
    customers = spark.createDataFrame([(i,'customer_'+str(i)) for i in range(8)],'customer_id LONG, name STRING')
    joined = orders.join(F.broadcast(customers),'customer_id')
    joined.explain(mode='formatted')
    assert joined.count()==1000
    salted = orders.withColumn('salt',F.pmod('order_id',F.lit(4)))
    partial = salted.groupBy('customer_id','salt').count()
    final = partial.groupBy('customer_id').agg(F.sum('count').alias('count'))
    assert counts.exceptAll(final).count()==0 and final.exceptAll(counts).count()==0
finally:
    spark.stop()

What each block does and the UI exercise answer

The range and arithmetic create reproducible data: customer 7 has 800 rows and customers 0–4 each have 40. Explain inspects planning; count triggers computation; the grouped result has six rows. The unique dimension preserves all 1,000 rows under the join. An explicitly broadcast eligible dimension should produce a broadcast strategy in the plan, but the final plan is what to inspect in an actual run. Salting then unsalting preserves the full result bag via exceptAll, not merely its row count.

The simple count may already benefit from partial aggregation, so salting can add overhead and need not improve duration. No task-time ratio is promised. To do the original UI exercise independently, enable an isolated UI bound to loopback under your environment's security policy, keep authorized event logs for a history server, and inspect the five categories above. A fast job may end before a browser opens. The answer to “feel skew and fix it” is to establish large per-task input/records and preserved output, then measure the change—not to assume salting makes max/median less than two.

Closure, retry and driver-memory exercise

An RDD closure is serialized work sent to executors, not a shared reference to the driver's Python variables. Mutating a captured counter inside a task does not reliably update a driver counter; aggregate values through supported distributed operations instead. Likewise, a task can run more than once after failure or speculation. Sending an external payment or increment from a task is not made exactly-once by Spark lineage. The safe consequence is to make side effects cooperative/idempotent or publish through an appropriate sink protocol [Spark RDD].

collect() returns all result rows to the driver. A million narrow primitive values and a million large strings have very different memory footprints; serialized input size is not a driver-memory budget. Prefer distributed output, or take/show when a bounded sample answers the question. They bound returned rows, not necessarily scan or shuffle work.

Worked oral trace: transformations record dependencies; an action requests a result; the scheduler schedules required stages and partition tasks. Four maps and three reducers can have twelve logical blocks, but the indexed sort-shuffle path uses one consolidated data file plus an index per map output, not twelve required data files. Re-execution may produce additional attempts. A 25× task-duration ratio warrants checking bytes/records, host health, GC and spill; there is no one-number proof of skew.

Tradeoffs and production reality

Shuffle versus broadcast versus bucketing: a shuffle supports large distributed inputs but can spill and skew; broadcast avoids the large-side repartition only when the build representation is bounded safely; compatible bucketed tables may avoid a repeated exchange if Spark recognizes the layout and join keys. Bucketing is not a universal promise that every later join is shuffle-free; verify physical plans, settings and storage support, and include rewrite/maintenance costs.

Skewed join case: 80 events with one key and two matching dimension rows must emit 160 pairs. Salting one side and replicating matching dimension rows across salts preserves those pairs; independently random-salting both sides loses matches, and replicating both sides multiplies them. AQE skew join may split supported large partitions, not arbitrarily split every operator. This explains the retained hot-user feature-engineering scenario without attributing an unsupported four-hour-to-25-minute result to Uber.

Table-format case: Iceberg partition evolution and snapshot commits address metadata/layout/commit problems; they are not the same feature as Spark's runtime AQE. A file format does not establish “70% of performance,” and a supposed 2019 Spark 3.0 result is chronologically suspect. Select a table format for required evolution/transactions, then inspect scans and exchanges separately.

Small-file case: increasing repartitions can reduce bytes per task but increase scheduling, metadata and output-file overhead. Coalesce/AQE may help in eligible stages; avoid coalesce(1) as a blanket cure because it can create a serial bottleneck. Use measured bytes, output requirements and compaction—not an invented per-task scheduling constant or an unsupported Airbnb incident.

Where this connects next

Kafka supplies retained input for Structured Streaming; the same relational API can express bounded or streaming computations with different supported operators. Airflow submits finite Spark applications and waits for completion; dbt adapters can generate Spark SQL. Snapshot table commits depend on the catalog/storage commit protocol, not a statement that Iceberg itself is a consensus algorithm. Offline feature pipelines still need historical correctness and online-serving contracts.

Recall and worked answers

1. Laziness? Transformations describe work; actions require results. Planning can precede actions and API metadata work can occur earlier.
2. Stage boundaries? Shuffle dependencies divide scheduler stages; branching, reuse, broadcasts, AQE and multiple jobs defeat the universal shuffles-plus-one shortcut.
3. Shuffle in one sentence? Redistribute records into downstream partitions, materializing/fetching logical blocks; N×M logical blocks do not imply N×M physical files or requests.
4. Skew diagnosis? Uneven input/work across partitions, corroborated by records/bytes and duration; duration ratio alone also fits GC/host/spill problems.
5. Salting? Events get one stable salt; matching dimension rows are replicated to every applicable salt; join on (key,salt). For aggregation, combine associative partial state—sum/count for means.
6. Broadcast choice? Supported join/build side, estimates/hints, threshold and runtime AQE rules; 10MiB is a default estimate threshold, not a memory guarantee.
7. collect danger? All result rows enter driver memory; it is safe only when the result is bounded for that budget. take/show bound returned rows, though computing them may still be costly.
8. AQE's three functions? Coalescing small shuffle partitions, supported skew-join handling and runtime join-strategy changes; not guaranteed speedups or a cure for all skew.
9. RDDs bypass Catalyst why? Arbitrary closures lack relational expression structure; use RDDs for justified custom algorithms/partition control, not because SQL is always faster or RDDs are forbidden.

Failure answer: doubling hash partitions does not split one hot key. Use an applicable salted/partial algorithm or supported adaptive join technique, preserving multiplicity and testing performance separately.

Explain out loud: trace one action's plan, distinguish physical files from logical shuffle blocks, and interpret a 25× duration ratio as an investigation starting point.

Sources and next step

Primary sources checked 2026-10-07: Spark 3.5.7 RDD guide, SQL tuning, index shuffle resolver at tag v3.5.7. Exact tagged-source passages, hashes and release evidence are retained.

Next: Kafka fundamentals. CPU correctness is established for the exact model; Spark runtime, UI, cluster failure and performance remain unexecuted, not missing conceptual teaching.

Original resource trail and reading purpose

The original optional pre-read destinations are retained below, not used as primary evidence for numerical claims, vendor incidents, or current API defaults. Video identity, duration, attribution and playback have not been independently verified here; search-result links are discovery links, not a particular verified lecture. Mutable latest documentation may differ from this lesson's pinned release. Use the versioned primary sources above for the governing contract.

• Apache Spark in 100 Seconds — original optional video/discovery destination; historical label, not verified attribution.
• A Deep Dive into Spark SQL's Catalyst Optimizer with Yin Huai — original optional video/discovery destination; historical label, not verified attribution.
• Reading the Spark UI — Jobs, Stages, Tasks — original optional video/discovery destination; historical label, not verified attribution.
• Spark — RDD Programming Guide — original optional reading destination; historical label, not verified attribution.
• Spark SQL Performance Tuning — original optional reading destination; historical label, not verified attribution.
• Adaptive Query Execution in Spark 3 — original optional reading destination; historical label, not verified attribution.

Read with a task: trace closures, actions, shuffle dependencies and initial/final plans; compare every tuning default with Spark 3.5.7 rather than the moving latest guide.

Series navigation: previous lesson; learning-plan hub. The original series used a 90-minute intuition → visual → hands-on → production → recall structure; that historical pacing is not a measured completion time for this revised lesson.