RDDs: The Core Abstraction

Under every DataFrame there is an RDD. RDD means Resilient Distributed Dataset. It is the oldest and lowest layer of Spark.

You will rarely write RDD code today. But the RDD explains three ideas that every later chapter uses: partitions, lineage and lazy work.

You will learn to:

1. Make an RDD from the shop orders and see its partitions.
2. Use lineage to explain how Spark recovers lost data.
3. Run transformations and actions on an RDD.
4. Choose between an RDD and a DataFrame.

An RDD is one list of items, cut into partitions. A partition is one piece of the list. Each task works on one partition, so many partitions can be worked on at the same time.

The SparkContext, sc, makes RDDs. parallelize() turns a Python list into an RDD, here with 3 partitions:

sc = spark.sparkContext

orders = [
    (101, 1, "paid", 40.00),
    (102, 1, "paid", 25.00),
    (103, 2, "refunded", 60.00),
    (104, 3, "paid", 15.50),
    (105, 3, "pending", None),
    (106, 4, "paid", 80.00),
]
rdd = sc.parallelize(orders, 3)

print(rdd.getNumPartitions())
print(rdd.map(lambda o: o[0]).glom().collect())

These are the same 3 partitions as the model in chapter 2. Partition 0 holds orders 101 and 102.

An RDD is immutable: you never change it. A transformation such as filter makes a new RDD, and the old one stays the same.

A transformation describes a new RDD, such as filter or map. An action asks for a result, such as collect, count or take.

Transformations are lazy: Spark only writes them down. Work starts at the action. Here the job count proves it:

paid = rdd.filter(lambda o: o[2] == "paid")
pairs = paid.map(lambda o: (o[1], o[3]))
# no job has run yet

print(pairs.collect())

A lambda is a short function written on one line. o[2] is the status of an order, and o[1] is its customer id.

reduceByKey adds up the values that share a key. Here it gives the paid revenue of each customer:

revenue = pairs.reduceByKey(lambda a, b: a + b)
print(sorted(revenue.collect()))

Other common actions are count(), which counts the items, and take(n), which returns the first n. On the paid orders, count() gives 4.

Lineage is the list of steps that built an RDD. Spark keeps it for every RDD. toDebugString() prints it:

print(revenue.toDebugString().decode())

(3) PythonRDD[7] at collect
 |  MapPartitionsRDD[6] at mapPartitions
 |  ShuffledRDD[5] at partitionBy
 +-(3) PairwiseRDD[4] at reduceByKey
    |  PythonRDD[3] at reduceByKey
    |  ParallelCollectionRDD[0] at readRDDFromFile

The +- line marks a shuffle. reduceByKey must bring all amounts of one customer together, so the work splits into two stages there (chapter 5).

Resilient means Spark can rebuild lost data. Suppose the machine that holds partition 1 stops.

Spark does not need a backup copy. It runs the same steps again on orders 103 and 104, on another machine.

This costs little before a shuffle, because one partition needs only its own input. After a shuffle, one lost partition can need input from every partition before it, so the rebuild costs more.

Lineage has a limit. If the source itself is gone, such as a temp file that was deleted, Spark cannot run the steps again.

Checkpointing helps here. It writes a copy of an RDD to safe storage, so recovery can start from that copy. Chapter 19, Caching, Persistence & Checkpointing, shows how.

A DataFrame is also an RDD underneath. You can see it with .rdd:

df = spark.read.csv("orders.csv", header=True, inferSchema=True)
print(df.rdd.first())

So why use DataFrames? A DataFrame has a schema, so Spark knows the columns and their types. Its optimizer, Catalyst, can then rewrite your query into a faster plan.

In an RDD, Spark sees only your lambda functions. It cannot look inside them, so it cannot improve them.

Use DataFrames and Spark SQL by default. Use an RDD only when the DataFrame API cannot do the job, for example to control partitions by hand.

Check yourself. Why can Spark rebuild partition 1 without a backup? (Answer: it keeps the lineage and runs the same steps again.)