Spark RDDs and Execution Model

Data Engineering & Modern Data Platforms · Lesson 29 · Big Data Processing and Apache Spark

Lesson Overview

The landing in PySpark Data Engineering is a DataFrame plan: a schema, a hash, a shuffle on event_id, a partition overwrite, a checkpoint. Under that plan Spark still schedules resilient distributed datasets. An RDD is a partitioned, immutable collection with a lineage: a recipe for rebuilding a lost partition from its parents. DataFrames are implemented on that idea, with a schema and an optimizer in front. You can ignore RDDs until a stage reruns, a Python lambda makes the rerun change the answer, or someone proposes to rewrite the landing as map and groupByKey because a blog called RDDs "the real API."

This lesson opens the execution model so those moments are readable. It reimplements the collision check as an RDD only to compare it with the DataFrame version, and then it stops. TechLearningPro does not ship the RDD job. The DataFrame job remains the landing. The RDD is how you see narrow versus wide dependencies without Catalyst hiding the exchange, how reduceByKey differs from groupByKey, how persist and checkpoint cut recomputation in different ways, and why a Python function on an RDD always crosses into a Python worker even when the same logic as a column expression would not.

The contract does not move. Identical canonical payloads collapse. Two payloads refuse the run. Prohibited fields refuse the file. completion-v3 stays downstream of the publish, as ETL and ELT Architecture placed it. Lineage is how a task retry recomputes. It is not a second business checkpoint. The business checkpoint is still the document lesson 28 writes after commit. Mixing those two words is the mistake this lesson is here to prevent.

Examples use PySpark 3.5 on local[2]. /learn/python is the lambda syntax. The lambdas are the thing the executor will pickle and ship. A lambda that closes over a large object ships that object. A lambda that calls time.time or reads a global counter is a lineage bug. Those are execution-model facts, not style nits.

Learning Objectives

By the end of this lesson you will be able to:

1. Define an RDD by its partitions, its compute function, and its dependency lineage.
2. Explain how a lost partition is recomputed, and which failures lineage cannot heal.
3. Distinguish narrow transformations from wide ones, and reduceByKey from groupByKey, on the collision check.
4. Separate persist, RDD checkpoint, and the landing's business checkpoint.
5. Say why a Python RDD pays a worker process for every transformation, while a DataFrame built-in may not.
6. Identify nondeterministic functions that make a retry publish different bytes.
7. Decide, for the progress landing, to keep the DataFrame implementation, and name a real reason to drop to RDDs in some other job.

Prerequisites

• Apache Spark Architecture: tasks, stages, shuffles, and executor loss. This lesson is the data structure those tasks recompute.
• Spark DataFrames and Spark SQL: the collision check you will restate without Catalyst. If the DataFrame version is not clear, the comparison will not mean anything.
• PySpark Data Engineering: the business checkpoint and the publish order. RDD checkpoint is a different tool. Do not merge them in your notes.
• Big Data Fundamentals: partitions as the unit of failure. An RDD partition is that unit with a recipe attached.
• Python 3.11 and PySpark 3.5 locally if you run the snippets. You should be comfortable reading a lambda. You do not need to have written Spark in Scala. The engine is still a JVM. Python RDDs are guests on it.

Detailed Core Concepts

What an RDD is

A resilient distributed dataset is a read-only, partitioned collection of records. Each partition has a function that can compute its records from parent partitions or from a source such as a file split. The graph of those functions is the lineage. The driver holds the graph. Executors compute partitions when an action forces them. Nothing is read when you call map. The same laziness as DataFrames is the original RDD laziness. DataFrames inherited it.

Records inside an RDD are opaque to the optimizer. A map that pulls event_id out of JSON looks, to Spark, like an arbitrary Python or JVM function. Spark cannot push a projection into Parquet through that function, because it cannot see the columns. It can still retry the function. Opacity and resilience are different properties. RDDs give you the second even when they withhold the first. That is why they remain the right tool for some byte-level work and the wrong default for the progress table.

sc.textFile builds an RDD of lines from a path, with partitions aligned to input splits, the same splits lesson 25 and lesson 26 discussed. A gzip member remains one partition. The RDD API does not repair a codec. parallelize builds an RDD from a local collection by shipping that collection from the driver. It is for fixtures and for tiny dimension lists. parallelize of the peak-day file would collect the day onto the driver first, which is the failure you already know, performed in the opposite order.

Narrow lineage and wide lineage

A narrow transformation computes each output partition from one parent partition, or from a fixed small set. map, filter, flatMap, and mapPartitions are narrow. They pipeline. A lost partition reruns the chain on the parent partition that still knows how to read its split. No shuffle data is required.

A wide transformation computes an output partition from many parent partitions. The mechanism is a shuffle. reduceByKey, groupByKey, join, distinct, and repartition are wide. The lineage of a wide step records the shuffle dependency. If a reduce task is lost before its output is consumed, Spark recomputes the map partitions that fed it, unless those shuffle files are still on disk or on an external shuffle service. Lesson 26's executor-loss story is this graph. The RDD is the graph you can print with toDebugString.

union is narrow in the sense that it concatenates partitions without a shuffle. coalesce with shuffle disabled can narrow partitions by merging neighbors. repartition is wide even if you are increasing the count only slightly. The words match the DataFrame lesson. The RDD API makes the shuffle harder to miss because you called a method whose name is the shuffle.

groupByKey and reduceByKey

Both group records by key. groupByKey ships every value to the reducer. The reducer sees an iterator. If one event_id was retried millions of times, one reducer holds millions of hashes, and the task dies. reduceByKey takes an associative, commutative function and combines values on the map side before the shuffle, then combines again on the reduce side. The collision function "two hashes become a conflict sentinel, two equal hashes stay that hash, a sentinel stays a sentinel" is associative. The shuffle carries one string per key per map partition, not every duplicate line. That is the same design intent as countDistinct on a hash in the DataFrame job: do not move the raw event.

A function that is not associative will silently compute the wrong answer under reduceByKey, because Spark is free to pair values in any order and to pre-aggregate. "Latest timestamp wins" is not a safe reduce unless you define a merge of pairs that actually is associative, and even then it is the wrong business rule. The sentinel merge is the rule that matches the contract. If you cannot write the rule as a reduce, you may need a different algorithm. You still do not need groupByKey for this one.

Actions force the graph

count, collect, take, saveAsTextFile, reduce, and foreach are actions. take(1) on a narrow RDD can compute a subset of partitions. take(1) after reduceByKey still needs the shuffle for the keys it materializes, and a conflict check that only takes one conflict still builds the grouped RDD. Do not promise an early exit in the run book. collect of the grouped conflicts is safe only because conflicts should be rare and the test fixture is small. Production returns a count of conflict keys or uses take(1) to prove non-emptiness, and writes the landing with the DataFrame writer, not saveAsTextFile of JSON lines.

foreach that writes to an external database from each task is an action with side effects and no committer. Retries double-write. The landing does not do this. Side effects belong in the driver after a small action, or in a committer-backed write.

Persist, checkpoint, and the business record

persist (and cache, which is persist at the memory level) materializes the RDD and keeps partitions on the executors according to a StorageLevel. MEMORY_AND_DISK spills partitions to local disk when memory is tight. The lineage is kept. If a persisted partition is lost, Spark recomputes it from lineage and can persist it again. Persist is an optimization across actions in one application. It is not durable across an application death. The driver died, the executors died, the cache died.

checkpoint on an RDD writes the materialized records to a reliable directory you set with setCheckpointDir, typically HDFS, and truncates the lineage. The parent graph can be discarded. Recovery reads the checkpointed data instead of rerunning back to the original file. That is useful for a very long iterative graph, where recomputing from the start is more expensive than reading a snapshot, or for a lineage that includes a nondeterministic step you unfortunately cannot remove and must not rerun. Checkpoint is itself an action-scale cost: it writes a full copy. It does not record your input hash, your reject cap, or your prohibited-field decision. It is not the landing checkpoint. Calling it by the same English word is unfortunate and traditional. In this course the landing document is the business checkpoint. The RDD method is a lineage cut. If you need both, they are two writes for two reasons, and only the business checkpoint authorizes completion-v3.

Unpersist when you are done. A long-lived session that checkpoints and persists every intermediate of a backfill will fill the checkpoint directory and the executor disks. The scheduled application should exit.

What lineage will not forgive

Lineage reruns functions. If the function returns a different answer the second time, the recomputed partition disagrees with partitions that did not fail. time.time, a random id, a read of a file that another job is appending, and a global counter are nondeterministic. The published date can change because one executor was preempted, with no change in the input. The landing's functions must be deterministic given the bytes. Canonical hashing of contract fields is deterministic. A Python UDF that "fills in" a missing event_time with the processing clock is not, and it is also a contract violation. Do not repair nondeterminism by checkpointing after it so the bad value freezes. Remove the clock.

Lineage cannot rerun if the source split is gone. Deleting the inbox file at the start of the job makes every retry fail. Keep the closed input until the business checkpoint is durable and you have decided retention, as lesson 28 already required.

Lineage cannot recreate the driver. The graph lived there. A new application builds a new graph. The business checkpoint is what makes the new application a skip or a replace rather than a duplicate append.

A task that fails deterministically, such as a parse that raises on a prohibited field, will fail again on every retry until spark.task.maxFailures ends the stage. Use that for breaches you intend to fail the stage on. Quarantine by emitting a tagged record, not by throwing, if the line is allowed to be bad under the cap. An exception is not a quarantine strategy. It is a stage-failure strategy.

Python RDDs and the worker

A Python map is serialized to the executors and run in Python worker processes. Every record crosses from the JVM, which read the split, into Python and back. There is no Catalyst step that turns the lambda into JVM bytecode. The DataFrame version of to_timestamp never enters that worker. The RDD version of the same parse always does, if you wrote it in Python. Scala or Java RDDs run inside the executor JVM and avoid that particular tax, and they still lack the optimizer. This course's language on the driver is Python. The landing stays on DataFrame column functions so the hot path stays in the engine. An RDD prototype is a teaching tool and a legitimate choice for a small opaque format the reader cannot express. It is not a performance upgrade for progress JSON that from_json already understands.

Closures are part of the model. A lambda that references a large dict from the driver ships a copy with the task. Broadcast the dict if many tasks need it and it is small enough. Do not close over the driver-side list of every event_id.

When an RDD is the right API

Use an RDD when the data is not a table and the operation is truly opaque: a custom binary framing you must split with mapPartitions, a compression scheme the file source does not implement, or a numerical kernel you are porting that already speaks iterators. Use an RDD when you need a particular Partitioner for a sequence of key-local steps and you have measured that the DataFrame shuffle does not express it. Use an RDD to read a legacy pipeline you must operate.

Use a DataFrame when the data has columns, when Parquet or JSON schemas matter, when you want predicate and projection pushdown, and when the landing's contract is a set of named fields. Progress events are that case. The comparison in the walkthrough should end with the DataFrame job still scheduled and the RDD version retained as a test that the sentinel merge matches countDistinct on a fixture. Two implementations in production will drift. One implementation in production, one oracle in a unit test, is enough.

Architecture and Design Diagrams

Lineage for the teaching RDD is a narrow parse and a wide reduce. A lost reduce task recomputes from the shuffle, or from the text splits if the shuffle files are gone.

mermaidCopy
flowchart TD
    file[Closed JSONL splits]
    lines[RDD of lines]
    pairs[Narrow map to event_id and hash or a breach tag]
    reduced[Wide reduceByKey with a conflict sentinel]
    action[Action: count sentinels]
    file --> lines --> pairs --> reduced --> action
    reduced -.->|lost task| pairs
    pairs -.->|shuffle files gone| lines

Persist keeps partitions and keeps lineage. Checkpoint writes records and cuts lineage. The business checkpoint is outside this graph.

mermaidCopy
flowchart LR
    rdd[RDD lineage]
    persist[persist on executors]
    ckptRdd[RDD checkpoint on durable storage]
    biz[Business checkpoint after the DataFrame commit]
    rdd --> persist
    rdd --> ckptRdd
    biz -.->|not a lineage cut| rdd

The production path stays the DataFrame landing. The RDD graph is the explanation underneath, not a second publish.

mermaidCopy
flowchart LR
    df[DataFrame landing publishes Parquet]
    rdd[RDD prototype in tests]
    mart[completion-v3]
    df --> mart
    rdd -.->|same fixture, not a second table| df

Technical Explanation

Printing the lineage

toDebugString on the driver shows the dependency types. A narrow step is indented under its parent. A shuffle is marked as such. You read it when the UI says a stage reran and you want the recipe, not the metrics. It is a string for humans and for a teaching assertion that the reduce is a shuffle dependency. It is not a stable API to parse in production control flow. Prefer the UI and the business checkpoint for operations. Prefer toDebugString while you are learning a plan or reviewing a prototype that does too much.

The sentinel must be a real reduce

Let a parsed good record be the pair (event_id, canonical_hash). Define:

pythonCopy
CONFLICT = "CONFLICT"

def merge_hash(left: str, right: str) -> str:
    if left == CONFLICT or right == CONFLICT or left != right:
        return CONFLICT
    return left

merge_hash("aa", "aa") is "aa". merge_hash("aa", "bb") is CONFLICT. merge_hash(CONFLICT, "aa") is CONFLICT. Any parenthesization yields the same result, so map-side combine is legal. A hash that equals the sentinel string would be a disaster. Use a sentinel that cannot be a hex digest, or a small tuple ("ok", hash) and ("conflict",) merged by a function that prefers conflict. The tuple form is safer:

pythonCopy
def merge_state(left: tuple, right: tuple) -> tuple:
    if left[0] == "conflict" or right[0] == "conflict" or left[1] != right[1]:
        return ("conflict",)
    return left

Each good line maps to ("ok", hash). The reduce never confuses a hash with a flag. This is the kind of care countDistinct hid inside the engine. When you drop to RDDs you take that care back.

groupByKey would let you write the same rule in a loop over the iterator. It is easier to read once and worse on a hot key. A teaching diff of the two plans belongs in a notebook. The shipped prototype, if you keep one at all, uses reduceByKey.

mapPartitions and quarantine

map calls the function per record and pays Python call overhead per line. mapPartitions calls it once per partition with an iterator, which is the right shape if you must use Python and you want to reuse a parser setup. It is still Python. It is still opaque. Inside the iterator you can yield either an accepted tuple or a reject tuple and tag them. You do not throw on a single bad line unless you intend to fail the task and therefore the stage after retries. Emitting a tagged reject is the RDD version of lesson 27's reject_reason. The driver counts tags with a narrow filter and a count, which is a shuffle-free action if the filter is narrow. The conflict reduceByKey is the action that shuffles.

Partition counts

textFile takes an optional minimum partition count. It cannot split unsplittable files, and it will not make more partitions than splits if the splits are fewer, except as the API's minimum interacts with the input format. Setting the minimum to 10,000 does not create useful parallelism on a 1 MB fixture. It creates scheduler noise. The DataFrame lesson's warning on spark.sql.shuffle.partitions has an RDD cousin: reduceByKey's partition count defaults from spark.default.parallelism, which in local mode is the thread count and on a cluster is often the total core count. A default of eight on a laptop and a default of two hundred on YARN are different shuffles. Set the number on reduceByKey(merge_state, numPartitions) explicitly in any prototype you care about, for the same reason lesson 28 sets files_per_date.

Comparison on a fixture, not on two tables

The useful test runs the DataFrame conflict_keys and the RDD reduce on the same lines and asserts they agree on which event_id values conflict. That test locks the sentinel to the SQL definition. It does not write a second Parquet dataset. The moment both write, an operator will point completion-v3 at the wrong one during an incident. One publish path.

Determinism drill

Include a negative example in the test file, fenced so nobody imports it into run: a map that appends time.time(). Force a recompute by counting twice without persist, or by calling unpersist and an action again. The two collects of a single line's timestamp differ. That is the demonstration. A landing function that fails this drill is not allowed near the publish. You do not need to fail a task in the test. You need the reader to see that lineage replays the function.

Practical Implementation or Design Walkthrough

Problem

Implement the collision sentinel with reduceByKey on a fixture, compare it to the DataFrame check, and show one nondeterministic map as a counterexample. Do not publish Parquet from the RDD.

Session

The RDD API hangs off the same SparkSession as the DataFrame job. Do not start a second context.

pythonCopy
from pyspark.sql import SparkSession

def session() -> SparkSession:
    return (
        SparkSession.builder.master("local[2]")
        .appName("tlp-rdd-comparison")
        .config("spark.ui.enabled", "false")
        .config("spark.sql.shuffle.partitions", "4")
        .config("spark.default.parallelism", "4")
        .config("spark.sql.session.timeZone", "UTC")
        .getOrCreate()
    )

Parse and reduce

The parser here is deliberately small and duplicated with lesson 27's intent on a fixture of already-normalized hashes. Production parsing stays in classify_lines. This function exists so the reduce is obvious.

pythonCopy
def conflict_ids(sc, pairs: list[tuple[str, str]], partitions: int = 4) -> set[str]:
    rdd = sc.parallelize(pairs, numSlices=partitions)
    states = rdd.map(lambda item: (item[0], ("ok", item[1])))
    reduced = states.reduceByKey(merge_state, numPartitions=partitions)
    conflicts = reduced.flatMap(
        lambda item: [item[0]] if item[1][0] == "conflict" else []
    )
    return set(conflicts.collect())

parallelize of a fixture list is acceptable. parallelize of a production extract is not. collect into a set is acceptable because the fixture's conflict count is tiny and the test asserts membership. Replace collect with count if you only need existence.

pythonCopy
def test_agrees_with_distinct_hashes() -> None:
    from pyspark.sql import functions as F

    spark = session()
    pairs = [("e1", "h1"), ("e1", "h1"), ("e2", "h2"), ("e2", "h9"), ("e2", "h2")]
    found = conflict_ids(spark.sparkContext, pairs)
    assert found == {"e2"}
    found_again = conflict_ids(spark.sparkContext, pairs, partitions=2)
    assert found_again == found
    frame = spark.createDataFrame(pairs, ["event_id", "canonical"])
    distinct = frame.groupBy("event_id").agg(
        F.countDistinct("canonical").alias("versions")
    )
    sql_conflicts = {
        row["event_id"] for row in distinct.where(F.col("versions") > 1).collect()
    }
    assert sql_conflicts == found
    # Teaching aid: the string shows a shuffle dependency. Do not parse it in main.
    debug = (
        spark.sparkContext.parallelize(pairs)
        .map(lambda item: (item[0], ("ok", item[1])))
        .reduceByKey(merge_state)
        .toDebugString()
    )
    if isinstance(debug, bytes):
        debug = debug.decode()
    assert "shuffle" in debug.lower()

The if False branch is noise. Do not ship it. The assertion that matters is sql_conflicts == found. toDebugString is printed in the teaching run so you can see the shuffle dependency. A cleaner test drops the print and the dead branch and only asserts the set equality. Use that cleaner test in the repository. The decode call depends on Spark returning bytes in some versions and strings in others. In PySpark 3.5, toDebugString returns a string. Call it directly. Do not decode a string.

A groupByKey variant on a synthetic hot key belongs in a comment or a separate benchmark you do not run in CI. Building millions of duplicates in a unit test wastes the suite. Explain the memory shape in prose, and if you measure it, measure it as a noted experiment with a cap.

A nondeterministic map you will not import

pythonCopy
import time

def unstable(sc):
    rdd = sc.parallelize(["line"], 1).map(lambda line: (line, time.time()))
    first = rdd.collect()
    second = rdd.collect()
    return first, second

Two actions, no persist, two clocks. The pair differs. An engineer who persist-once and then checkpoints has frozen one arbitrary clock into a durable file and called it recovery. The landing's answer is to delete time.time from the parse. Processing time, if you need it for operations, is a column the driver adds once to the checkpoint document, not a column on each event. Event time remains the player's timestamp.

What the prototype must not grow into

Do not add saveAsTextFile to the publish root. Do not add a second checkpoint directory that operators might confuse with checkpoint_path in BatchConfig. Do not call sc.setCheckpointDir on the dataset root. An RDD checkpoint writes part files that are not the Parquet contract. A curious SQL job pointed at the dataset root will fail or, worse, succeed with a schema you did not review. If you experiment with checkpoint, use a scratch prefix and delete it in the test's teardown.

How a lost task replays the sentinel

Narrate one failure. The reduce task for a subset of keys dies after the map tasks wrote shuffle files. The driver reschedules the reduce task. It reads the same shuffle files. merge_state runs again. Because it is deterministic and associative, the sentinel for e2 is the same. If the shuffle files were lost with the executor, the map side reruns lambda item: (item[0], ("ok", item[1])) on the parent partitions. Those parents, in the fixture, are parallelize, which still has the list in the driver. In production they would be the text splits of the closed file, which must still be there. The DataFrame landing's recompute is the same story with a less visible graph. When the UI shows a map stage that ran twice, this is the mechanism, not a double publish. The publish happens once, in the driver, after the action returns, in the DataFrame job. The RDD prototype returns a set to the test.

Enterprise Real-World Scenarios

groupByKey in a hot retry storm

A prototype of the collision check used groupByKey because the loop was obvious in review: for each event_id, look at the set of hashes, fail if the set is larger than one. On a quiet fixture the stage finished. During a client outage the player retried a single lesson checkpoint a very large number of times for a handful of event_id values. The planning assumption of fifty million events was not the problem. The skew was. One reduce task held the iterator of that hot key and was killed for memory. The task retried on another executor and died the same way. The stage failed. The DataFrame landing, which had been the scheduled job, was not the code that failed. An engineer had swapped in the RDD prototype for the day's run because it was "easier to debug." The business checkpoint had not moved. The mart was stale, correctly, and the postmortem had to explain an unscheduled API.

The repair was to delete the prototype from the submit path and keep it as a unit test that only parallelizes a handful of pairs. The scheduled job stayed conflict_keys on the DataFrame. A note in the run book says a hot key is a producer incident: the DataFrame job still has a skew problem if one key's hash state does not fit, but countDistinct on a single hash value per duplicate is far smaller than a list of every raw line, and the cap on copies belongs in the producer conversation. reduceByKey with the sentinel would have survived where groupByKey died, and it would still have been the wrong process to run at 01:00 UTC because it was not the tested publish path. Two correct collision algorithms and one publish path. The incident was the second path.

The security note was small and real. The groupByKey debug session collected the hot key's values to the driver to inspect them, and the values were full JSON lines in that prototype. Those lines entered the driver log. The DataFrame job's quarantine does not store the line. The prototype must not either, even in a notebook, on production data. Inspect counts and hashes. If you must see a line, do it in the raw zone's break-glass path, not in a collect.

A processing-time column that changed on retry

Another change added processed_at with a Python map over the DataFrame's underlying RDD, writing the result beside Parquet from a side path an analyst found. An executor preemption recomputed some partitions. Those partitions received a later processed_at. A downstream check compared file hashes between a rerun and the original and reported corruption. The input hash in the business checkpoint matched. The Parquet did not, because the RDD map was nondeterministic and the side path had been built with saveAsTextFile in append-like directory layout that a retry added files to. The analyst was right that the bytes changed and wrong that the player events had changed.

The repair removed processed_at from the event records. The business checkpoint already carries committed_at once, written by the driver, if you choose to add that field to the JSON document. It is one clock, recorded after success, not a clock per row. The side path was deleted so completion-v3 could not see it. A regression test calls the unstable function on a fixture and asserts the two collects differ, with a comment that this is why the column does not exist. Lineage did what it is designed to do. The function was not safe to replay. No storage replication setting would have fixed a function that disagrees with itself.

Best Practices

• Keep the scheduled landing on the DataFrame job. Use RDDs to understand lineage and to test the sentinel against countDistinct.
• Prefer reduceByKey with an associative merge over groupByKey when the value can be combined.
• Make every function in the lineage deterministic. Put clocks in the business checkpoint, once.
• Treat RDD checkpoint as a lineage cut for long graphs, and the JSON checkpoint as the only record the SQL task reads.
• persist only across actions you will actually rerun inside one application, and unpersist.
• Set numPartitions on reduceByKey explicitly in any prototype that leaves your laptop.
• Emit reject tags. Do not throw for a single quarantinable line.
• Do not parallelize production datasets from the driver.
• Do not saveAsTextFile into the publish root.
• Print toDebugString when you are learning a rerun. Do not parse it in main.
• Close over small objects. Broadcast a small map. Do not close over the day's keys.
• One publish path. A second implementation belongs in a test that compares sets on a fixture.

Common Mistakes and Anti-Patterns

Rewriting the landing as RDDs for control. You lose schema pushdown and you gain Python workers on every row, and you now maintain two contracts. The repair is to stay on classify_lines and read toDebugString or the UI when you need the graph.

groupByKey of raw lines. Hot keys collect the day into one task and die. The repair is a combine that shuffles a hash state, which the DataFrame aggregate already does.

time.time or uuid inside map. Retries change bytes that the business checkpoint still describes as the same input. Downstream hash checks fail. The repair is a pure function of the input bytes, and a single driver-side timestamp on the checkpoint if operations need one.

Using RDD checkpoint as the landing checkpoint. The SQL task does not know the input hash, and part files appear in a directory that is easy to misread as the dataset. The repair is lesson 28's document, written after the Parquet commit. Use RDD checkpoint only to cut a long lineage you have justified.

Throwing on bad lines. The stage fails after four identical retries, and the cap logic never runs. The repair is a tagged reject and a driver-side count, matching the DataFrame reason column.

collect of a production RDD to dedupe in a set. The driver holds the day, which is the lesson 24 failure with extra steps. The repair is an action that returns a count or a small conflict sample, and a writer that is not the driver looping on rows.

saveAsTextFile for the publish. Text has no schema, the commit is a directory of part files that later jobs misread, and append-style retries duplicate lines. The repair is the Parquet dynamic overwrite.

Parsing toDebugString to decide success. The string changes between versions. The repair is assertions on conflict ids in tests and the business checkpoint in production.

A second SparkContext. SparkSession already has one. A second context in the same process throws or isolates nothing useful. The repair is one session per application, which the architecture lesson already required.

Troubleshooting and Design Review

Symptom: A map stage runs twice and the output row count is stable, but a processed_at column moved.

Likely cause: executor loss recomputed a nondeterministic map. Lineage worked. The function did not.

Investigation: the map closure for clocks, random, and uncached external reads. Event log for executor loss.

Corrective action: remove the column. Record one timestamp on the business checkpoint if you need processing time. Republish the date from the closed input so the Parquet matches the hash.

Symptom: The collision stage dies on memory on one task, and the code path is an RDD.

Likely cause: groupByKey, or a reduce that still ships raw JSON.

Investigation: toDebugString for a shuffle dependency, and the serialized value. UI task metrics for one large task.

Corrective action: stop running the prototype. If the scheduled DataFrame job shows the same skew, project to hashes and cap pathological duplicate counts as a producer incident. Do not switch the submit path to RDDs to "see the records."

Symptom: After a successful RDD checkpoint experiment, SQL against the dataset root fails with a schema error.

Likely cause: checkpoint part files were written under the publish root or a child of it.

Investigation: list the root. Checkpoint files are not Parquet.

Corrective action: delete the scratch prefix, point setCheckpointDir at scratch only, and confirm the reader path. The business checkpoint file should still be outside the root, as JSON, not as an RDD checkpoint.

Symptom: reduceByKey results change when you change parallelism.

Likely cause: the merge function is not associative, so different pairings yield different answers. "First one wins" depends on order.

Investigation: a fixture with three hashes and two partition counts.

Corrective action: use the sentinel merge or return to countDistinct, which has a defined aggregate. Do not paper over the change by pinning parallelism and hoping the order stays.

Symptom: Python RDD stages dominate a job that used to finish as a DataFrame.

Likely cause: someone converted to df.rdd.map for a small convenience and forced the whole stage through Python workers.

Investigation: the plan and the code for .rdd.

Corrective action: move the logic back to column functions. Keep the RDD only where the data is opaque. Compare stage time on a slice before you accept the conversion.

Symptom: A retry cannot recompute because the source was deleted at job start.

Likely cause: a cleanup task raced the application. Lineage points at a missing split.

Investigation: the path in the exception and the retention job's schedule.

Corrective action: retain the closed input until the business checkpoint exists. Do not delete inputs in the first task. A missing source is not a Spark bug. It is a lifecycle bug.

A design review asks: What is recomputed if this reduce task dies? Is every function in that recipe deterministic? Are we shipping groupByKey or an associative reduce, and are we shipping it at all? Where is the business checkpoint relative to any RDD checkpoint directory? What does the driver collect? Does any path besides the DataFrame writer publish? Can a test show the sentinel and countDistinct agree? The review ends with the scheduled job unchanged unless a column-level bug was found.

Interview Questions and Answers

Q1. What makes an RDD resilient?

Each partition can be recomputed from its lineage: parent partitions and a deterministic function, down to a source split that still exists. If a task or an executor is lost, the driver reschedules that partition. The driver does not hold the rows. Resilience is not replication of the result. HDFS replication is a storage property. RDD lineage is a compute property. I still write the business checkpoint after a successful publish, because lineage does not survive the death of the driver and does not tell completion-v3 which input hash is current.

Q2. How is a DataFrame related to an RDD?

A DataFrame is a structured dataset with a schema and a Catalyst plan. Its execution still breaks into stages and tasks, and you can reach an RDD with .rdd if you accept opacity and, in Python, the worker boundary. I use the DataFrame for progress events so the engine sees columns. I drop to an RDD when the bytes are not a table. I do not maintain two production landings. I do keep a fixture where an RDD sentinel and countDistinct name the same conflicting event_id.

Q3. What is the difference between a narrow and a wide dependency?

A narrow dependency computes an output partition from one parent partition, like map and filter on a line. A wide dependency needs a shuffle, like reduceByKey on event_id, because duplicates can sit in any split. Narrow steps pipeline and recompute from the parent split. Wide steps recompute from shuffle files if those files remain, and from the map lineage if they do not. The stage boundary in the UI is this wide edge. toDebugString shows it while you are learning. The DataFrame explain shows the same edge as an exchange.

Q4. Why is groupByKey the wrong collision implementation?

It transfers every value for a key to one reducer. A retry storm on one event_id builds a huge iterator and the task runs out of memory, then fails the same way on retry. reduceByKey with an associative sentinel combines hashes on the map side and shuffles a small state. The DataFrame countDistinct on canonical is the version I schedule, for the same reason: the shuffle record is a hash, not the JSON line. I will not "fix" a hot key by collecting it to the driver.

Q5. What properties must the reduce function have?

Associativity and commutativity, because Spark combines values in any order and may combine them before the shuffle. The conflict sentinel has those properties: any disagreement or any existing conflict stays a conflict, and equal hashes stay that hash. "Latest row" does not, unless you carefully merge pairs and even then it violates the business contract. If reduceByKey answers change when numPartitions changes, the function is wrong. I test two partition counts on three values.

Q6. What is the difference between persist and checkpoint on an RDD?

persist stores partitions on executors, optionally spilling to local disk, and keeps lineage so a lost block can be rebuilt. It dies with the application. checkpoint writes the records to reliable storage and cuts the lineage so recovery does not walk back to the source. Checkpoint is a full extra write. Neither one records the input hash or authorizes the mart. The business checkpoint in the landing JSON is the only one the SQL task reads. I do not point setCheckpointDir at the Parquet root.

Q7. A rerun after preemption changed a column. The input hash did not. What happened?

A function in the lineage was nondeterministic. Lineage replayed it on the lost partitions only, so those partitions received new time.time or random values. The source bytes were the same, which is why the business hash matched, and the output bytes differed, which is why an analyst called it corruption. I remove the function. Processing time belongs once on the driver-side checkpoint, not on each event. I republish from the closed input.

Q8. Why are Python RDDs a poor default for this landing?

Every Python transformation runs in a worker process, with serialization per record, and the optimizer cannot see inside the lambda. Built-in DataFrame functions stay in the executor JVM and can be compiled into the stage. The landing's parse, hash, and filter are expressible as those functions. I would pay for Python RDDs only for an opaque format I cannot read otherwise, and I would measure mapPartitions against a real slice before I called it an improvement. Convenience is not a measurement.

Q9. What does lineage not recover?

A dead driver and its graph. A deleted source split. A deterministic exception you coded, which will throw again until the stage fails. A nondeterministic function's previous answer. An external side effect the task already performed, such as an uncommitted write to a database in foreach. I design the landing so the only durable side effect is the committer-backed Parquet write followed by the business checkpoint, both driven by the DataFrame job. Tasks stay pure.

Q10. When would you choose an RDD in a TechLearningPro job?

For a binary or custom-framed extract that has no schema I can hand to from_json or the Parquet reader, or for a legacy pipeline I must operate, or for a partitioner-sensitive algorithm I have shown the DataFrame API does not express. Not for the progress landing, not for completion, and not because RDDs feel closer to the metal. If I drop down, I keep the publish in one place and I keep the functions deterministic. A teaching prototype uses parallelize on a fixture and does not get a scheduler entry.

Q11. How do you confirm the RDD sentinel matches the DataFrame rule?

I run both on the same list of (event_id, hash) pairs, including an identical retry and a two-hash id. The sets of conflicting ids must be equal. I include a case where a key appears three times with two hashes so a non-associative bug has somewhere to hide, and I run at two partition counts. I do not compare two published datasets in the lake. The DataFrame path is the only publisher. The test lives with the unit tests that seed email in lesson 28.

Q12. An interviewer asks you to debug a stage that reran. What do you look at?

The event log for executor loss or speculation, the shuffle read metrics, and whether the source path still exists. I ask whether any function in that stage can return a different value if replayed. I look for groupByKey, .rdd, and time in the closure if the output changed. I do not look at the business checkpoint for evidence that a task reran. The checkpoint moves once per successful application. A stage rerun inside a successful application is lineage, and it should be invisible in the published bytes. If it is visible, the function is the defect.

Hands-On Exercises

Exercise: Compare one collision rule on two APIs

Problem statement. Implement the associative conflict sentinel on an RDD. Show that it matches a DataFrame countDistinct on a fixture. Show a nondeterministic map as a counterexample you will not ship. Leave the Parquet landing on the DataFrame path.

Tasks.

1. Define merge_state so equal hashes stay and any disagreement becomes a conflict. State why it is associative.
2. Build the pairs with parallelize on a small list. Do not read the production inbox.
3. reduceByKey with an explicit partition count. Assert the conflicting ids.
4. Compute the same ids with a DataFrame aggregate and assert equality. Repeat at a second partition count.
5. Implement groupByKey only as a short explanation of the memory failure, not as the scheduled job. You may skip executing a hot-key benchmark.
6. Demonstrate two collects of a time.time map differ. Say where a single processing timestamp is allowed to live.
7. Write the recompute story for a lost reduce task when shuffle files remain and when they do not.
8. State, in a paragraph, why this RDD does not write the publish root and does not replace the business checkpoint.

Expected deliverables. The test module, the counterexample, and the recompute paragraph.

Evaluation rubric.

You pass when a reviewer can say what recomputes after an executor loss, why the published bytes do not change, and which API the scheduler runs at 01:00 UTC.

Lesson Summary

An RDD is a partitioned lineage: a recipe the driver keeps and the executors run when an action forces them. Narrow steps recompute from a parent split. Wide steps shuffle. reduceByKey with an associative conflict sentinel matches the landing's collision rule on a fixture and avoids groupByKey's hot iterator. persist speeds a second action in the same application. RDD checkpoint cuts a long lineage. Neither replaces the business checkpoint written after the Parquet commit. Python RDDs pay a worker on every row. Nondeterministic functions make retries disagree with themselves. The progress landing stays the DataFrame job from lesson 28. The RDD model is how you read that job's retries and how you refuse a second publish path.

Spark Joins and Data Shuffling is next. The shuffle you saw here as reduceByKey becomes the join and the skew problem on a full progress fact, still published once, still under the same contract.

Further Reading

• RDD programming guide — partitions, transformations, actions, persist levels, checkpoint, and shuffle. The guide is the source for the execution model in this lesson.
• PySpark RDD API — textFile, reduceByKey, toDebugString, StorageLevel. Confirm return types on your patch before decoding strings.
• Spark cluster overview — how lost executors meet the scheduler that replays lineage.
• Matei Zaharia and colleagues, "Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing," NSDI 2012 — the paper that defines lineage. Read it for the model, then return to the DataFrame landing for the product.
• Spark SQL guide — the API that should remain on the submit path.
• Continue this course: Spark Joins and Data Shuffling. The business landing remains PySpark Data Engineering.