What is an RDD and why immutability matters

• Resilient – lost partitions are rebuilt from lineage (the recorded chain of transformations) rather than from stored replicas.
• Distributed – records are split into partitions that live across the executors of a cluster.
• Dataset – it is a typed collection of records (Java objects, Python objects, or rows).
• Immutable – every transformation returns a new RDD; the original is never modified, which eliminates race conditions in parallel execution.
• Lazily evaluated – transformations only build a plan; execution happens when an action like count or collect is called.
• Foundation – DataFrames and Datasets are built on top of RDDs, adding schema and Catalyst optimization.

What is an RDD?

An RDD (Resilient Distributed Dataset) is Spark's original core abstraction. It is an immutable, partitioned collection of records distributed across the executors of a cluster. You can think of it as a read-only array that is too large to fit on one machine, so Spark splits it into partitions and spreads them across the cluster.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("NBI-RDD-Demo").getOrCreate()
sc = spark.sparkContext

# Create an RDD from a list of account balances
balances = sc.parallelize(
    [("SAV-1001", 45000.00), ("CUR-2001", 120000.00),
     ("SAV-1002", 8500.00), ("CUR-2002", 340000.00)],
    numSlices=2
)

print(f"Partitions: {balances.getNumPartitions()}")
print(f"Type: {type(balances)}")

Partitions: 2
Type: <class 'pyspark.rdd.RDD'>

Why is immutability important?

Immutability means that once an RDD is created, its contents cannot be changed. Every transformation like filter or map returns a brand-new RDD, leaving the parent intact. This matters for three concrete reasons.

First, thread safety. Because partitions are read-only, many tasks can process them in parallel across cores and machines with no locks and no race conditions. Second, deterministic recomputation. If a partition is lost, Spark replays the exact same transformations on the exact same parent data and gets the exact same result. Third, a clean functional programming style where transformations are pure functions.

# Each transformation creates a NEW RDD, never modifying the original.
original = sc.parallelize([100, 200, 300, 400, 500])

# filter returns a new RDD; 'original' is unchanged.
high_value = original.filter(lambda x: x > 250)

print(f"Original count: {original.count()}")   # still 5
print(f"Filtered count: {high_value.count()}")  # 3
print(f"Same object? {original is high_value}")  # False

Original count: 5
Filtered count: 3
Same object? False

How does lazy evaluation work with RDDs?

Transformations (map, filter, flatMap, groupByKey) are lazy. They build a lineage graph that records what to compute but do not execute anything. Only actions (count, collect, saveAsTextFile, reduce) trigger actual computation.

This laziness lets Spark see the entire chain of operations before running, which enables optimizations like pipelining narrow transformations into a single stage.

# All transformations below are lazy: nothing runs yet.
txn_rdd = sc.textFile("s3://nbi-data-lake/raw-transactions/")
parsed = txn_rdd.map(lambda line: line.split(","))
settled = parsed.filter(lambda fields: fields[3] == "SETTLED")
amounts = settled.map(lambda fields: float(fields[2]))

# Check the lineage (debug string)
print(amounts.toDebugString().decode("utf-8"))

# NOW trigger execution with an action.
total = amounts.reduce(lambda a, b: a + b)
print(f"Total settled amount: {total}")

(4) PythonRDD[4] at reduce
 |  PythonRDD[3] at map
 |  PythonRDD[2] at filter
 |  PythonRDD[1] at map
 |  s3://nbi-data-lake/raw-transactions/ MapPartitionsRDD[0]
Total settled amount: 1284500000.50

How does fault tolerance through lineage work?

Spark records the directed acyclic graph (DAG) of transformations that produced each RDD. If an executor dies, Spark does not fail the job. It finds the lineage for the lost partitions and recomputes only those partitions from the parent data.

This is cheaper than replicating every intermediate result. At the bank, a failed node during a nightly job causes a small recompute rather than a full restart.

# Lineage enables automatic recovery.
# If the executor holding partition 2 of 'amounts' dies,
# Spark replays: textFile → map → filter → map
# ONLY for partition 2, not the entire dataset.

# You can see the lineage with toDebugString.
print(amounts.toDebugString().decode("utf-8"))

How is an RDD different from a DataFrame?

An RDD is untyped (from Spark's perspective) and not optimized by Catalyst. A DataFrame adds a schema (column names and types) and benefits from the Catalyst optimizer and Tungsten execution engine. DataFrames are typically 2x to 10x faster than equivalent RDD operations because of these optimizations.

| Aspect | RDD | DataFrame |
|---|---|---|
| Schema | No schema; records are opaque objects | Named, typed columns |
| Optimization | No Catalyst; runs your code as-is | Catalyst optimizer + Tungsten codegen |
| API | Functional: map, filter, reduce | Declarative: select, filter, groupBy |
| Type safety | Compile-time in Scala; runtime in Python | Runtime column checks |
| Performance | Slower (no optimization) | Faster (optimized plans) |
| Use case | Custom partitioning, low-level control | Standard analytics, SQL queries |

from pyspark.sql import functions as F

# RDD approach: manual, no optimization
rdd_result = (sc.textFile("s3://nbi-data-lake/transactions/")
    .map(lambda l: l.split(","))
    .filter(lambda f: f[3] == "SETTLED")
    .map(lambda f: (f[1], float(f[2])))
    .reduceByKey(lambda a, b: a + b)
    .collect())

# DataFrame approach: optimized by Catalyst
df_result = (spark.read.parquet("s3://nbi-data-lake/transactions/")
    .filter(F.col("status") == "SETTLED")
    .groupBy("accountId")
    .agg(F.sum("amount").alias("total"))
    .collect())

# The DataFrame version is much faster because Catalyst pushes
# filters to the Parquet reader and uses whole-stage codegen.

What are the rules and limitations of RDDs?

RDDs have several constraints that follow from immutability and distributed execution.

• You cannot modify an RDD in place. Every operation returns a new RDD.
• Operations that require seeing all data for a key (like groupByKey or join) trigger a shuffle.
• Closures sent to executors must be serializable. If you reference a non-serializable object, you get a NotSerializableException.
• Accumulators and broadcast variables are the only sanctioned way to share mutable state across tasks.

# Common mistake: referencing a non-serializable object in a closure
class BankConfig:
    def __init__(self):
        self.min_amount = 1000

config = BankConfig()

# This works because Python pickles the object.
# In Scala/Java, you must ensure Serializable.
filtered = balances.filter(lambda x: x[1] > config.min_amount)
print(filtered.collect())

[('CUR-2001', 120000.0), ('SAV-1001', 45000.0), ('CUR-2002', 340000.0)]

What about checkpointing for long lineage chains?

Very long lineage chains can make recomputation expensive. Checkpointing writes an RDD to reliable storage (like HDFS) and truncates the lineage so recovery starts from the checkpoint rather than replaying hundreds of steps.

This is essential for iterative algorithms like training a risk model over many iterations, where each iteration adds to the lineage.

# Set the checkpoint directory
sc.setCheckpointDir("s3://nbi-data-lake/checkpoints/")

# Build a long lineage through iteration
rdd = sc.parallelize(range(1000000), 100)
for i in range(50):
    rdd = rdd.map(lambda x: x + 1)
    if i % 10 == 0:
        rdd.checkpoint()  # Truncate lineage every 10 iterations
        rdd.count()        # Force materialization

# Without checkpointing, a failure at iteration 50 would
# replay all 50 map operations. With it, recovery starts
# from the last checkpoint at most 10 iterations back.

What should a senior engineer know about RDDs?

A senior engineer understands that DataFrames are the standard API, but RDDs still matter in specific cases: custom partitioning, low-level control over exactly how data is distributed, and interfacing with legacy code.

They also know that every DataFrame operation translates to RDD operations under the hood, so understanding RDD partitioning and lineage helps debug performance issues even when using the DataFrame API.

# Access the underlying RDD from a DataFrame
df = spark.read.parquet("s3://nbi-data-lake/accounts/")
underlying_rdd = df.rdd

print(f"Partitions: {underlying_rdd.getNumPartitions()}")
print(f"Partition sizes: {underlying_rdd.glom().map(len).collect()[:5]}")

Partitions: 200
Partition sizes: [4500, 4520, 4480, 4510, 4490]