Spark: RDDs, transformations and actions

One engine instead of three stacks

What is big data used for (p. 1154)? Reports track business processes; diagnosis asks why engagement drops or the system is slow, and detects spam, worms and DDoS attacks; decisions pick a treatment, a product feature, an ad. The slide's yellow box: "Data is only as useful as the decisions it enables".

Faster and better decisions set three processing goals (p. 1155): low-latency interactive queries on historical data ("identify why a site is slow and fix it"); low-latency queries on live data, streaming ("a worm may infect 1mil hosts in 1.3sec"); and sophisticated processing such as anomaly detection — iterative algorithms, machine learning, graphs. The next slide draws batch, interactive and streaming as the corners of a triangle around the words "Single Stack!": "Support batch, streaming, and interactive computations… … and make it easy to compose them" (p. 1156).

Why compose them? The DDoS example of p. 1158 needs all three at once. To decide whether a traffic surge is an attack you must "Detect attack pattern in real time" (streaming), ask "Is traffic surge expected?" (an interactive query on history), and that query is fast only thanks to "pre-computation (batch)" — "And need to implement complex algos (e.g., ML)!"

The slide's "Today's state-of-art analytics stack" (p. 1157) takes three systems: a demultiplexer copies the logs into a streaming stack (Storm, lesson 33) and a batch stack (Hadoop, lesson 31), which feeds an interactive store (HBase, Impala, lesson 32). The price, as the slide lists it: three stacks to maintain, "Expensive and complex"; "Hard to compute consistent metrics across stacks" — the same "errors per minute" coded twice; and "Hard and slow to share data across stacks", since the data moves between them as files.

Spark's answer is one general engine with libraries on top, in a stack of three layers (p. 1159): data processing, resource management, storage. The Berkeley Data Analytics Stack (p. 1161) fills them with Spark and its libraries, Mesos to share the machines, and HDFS or S3 under Tachyon, an in-memory storage layer. Spark owns only the top layer: "You can think of Spark as scalable computation platform", one that uses data stored in Cassandra, AWS S3, HDFS "And more" (p. 1201). It keeps nothing permanently: its input and its results live in those systems. Table 1 lines up the three drawings of the stack: the names change, the layers do not.

As the slides tell it (p. 1194), Spark began in 2009 as a research project of Matei Zaharia at the UC Berkeley AMPLab, went to the Apache Software Foundation in 2013, became a top-level Apache project in February 2014, and in November 2014 set a world record in large-scale sorting for Databricks, Zaharia's company.

It promised to be "Fast: up to 100x faster than Hadoop", "Easy to use: 5-10x less code than Hadoop" and "General: support interactive & iterative apps" (p. 1195). The 100x has fine print. Two slides later the same kind of claim, made for Shark against Hive, comes in two sizes: "Up to 100x when input is in memory", "Up to 5-10x when input is on disk" (p. 1197). Fig. 2 shows where the numbers come from.

Why is the iterative bar the most dramatic? An iterative algorithm reads the same data at every pass. The logistic regression of p. 1199 — the gradient descent of machine learning — computes one gradient term per point with points.map(...), sums the terms with .reduce(_ + _) and repeats. On Hadoop each pass is a new MapReduce job that reads its input from HDFS and writes its output back; Spark reads the points once, keeps them with features.cache(), and every further pass runs in memory. The features come from a SQL join on the same machines: the libraries "All share the same set of workers and caches".

The libraries are the old stacks rebuilt on the engine. Spark Streaming will "Implement streaming as a sequence of <1s jobs" (p. 1196): small batches, each an ordinary Spark job, so the stream inherits fault tolerance and straggler handling and counts every record exactly once (the slide's "exactly one semantics"; its last line prints "batch" twice where it means batch, interactive and streaming). The slide titled SparkSQL describes Shark, "Hive over Spark: full support for HQL and UDFs" — its box is the highlighted one; Spark SQL replaced it (lesson 40). And each layer speaks an existing interface (p. 1200): Kafka, Flume, Twitter and TCP sockets feed Spark Streaming; Shark keeps the Hive API, GraphX the GraphLab API; Mesos also runs Hadoop, Storm and MPI; Tachyon offers the HDFS API.

Spark is a computation engine, not a store: one engine and its libraries replace three stacks, and the data stays in the workers' memory between the steps.

RDDs: a recipe, partitions and a cache

Here is the job of p. 1222, the one from the opening, as you would type it today. It will "Load error messages from a log into memory, then interactively search for various patterns":

# sc is the SparkContext of your program (sc = spark.sparkContext)
lines = sc.textFile("hdfs://...")
errors = lines.filter(lambda s: s.startswith("ERROR"))
messages = errors.map(lambda s: s.split("\t")[2])
messages.cache()
# two actions: each returns one number to the driver
messages.filter(lambda s: "mysql" in s).count()
messages.filter(lambda s: "php" in s).count()

Read it line by line. textFile returns an RDD of lines, one partition per HDFS block — and reads nothing. filter returns a new RDD of the error lines, map one that keeps the third tab-separated field: still nothing runs. cache() only marks messages to be kept in memory once computed. The first count() starts the work (Fig. 3): one task per block; each worker reads its block, filters, maps, caches its partition of messages and counts the "mysql" lines; the three numbers are added up at the driver. The second count() starts from the caches: no disk, no filter, no map.

"Reads nothing" is literal. On PySpark 4.2.0, textFile on a path that does not exist, followed by a filter and a map, raises no error at all; "Input path does not exist" arrives only with count().

The slides give it "4 main features" (p. 1205): "Distributed Collection of Data", "Fault-tolerant", "Parallel operation - partioned" and "Ability to use many data sources"; and three adjectives: "immutable, lazily evaluated, cacheable" (p. 1208). Each word of the name is a mechanism. Resilient: a lost partition is recomputed, not restored from a copy. Distributed: the partitions sit on many machines and are processed in parallel, one task each. Dataset: you program it as one collection.

Its operations come in two kinds (p. 1209): "Transformations are a recipe to follow. Actions perform what the recipe says to do and returns something back." A transformation builds a new RDD "through deterministic operations on other RDDs" — map, filter, join — and is a "lazy operation"; an action will "return value or export data" — count, collect, save — and "triggers execution" (p. 1211). Table 2 collects the lists of the slides.

Why wait? "With a large dataset, you don't want to calculate all the transformations until you are sure you want to perform them!" (p. 1210). Laziness pays three times: nothing is computed that no action needs; consecutive transformations such as filter and map run as one pass over each partition, record by record, without storing the RDD in between; and Spark sees the whole plan before it runs anything, so it can cut it into stages and send the work to where the data is.

"RDD are immutable, transformations create new RDD" (p. 1213). So an RDD stays what it was born as — a parent and a function — and the chain of those, back to the file, is its lineage (Fig. 4). Spark will use the lineage to rebuild lost data, so a transformation must give the same result every time it runs: a function that drew random numbers or read the clock would rebuild different data. Immutability pays a second time in the cache: since nobody can change a cached partition, many jobs can read it at once without locks.

Transformations write the recipe, actions cook it: until an action, an RDD is a lineage of operations, not data.

Narrow and wide: does the data have to move?

Take the readings of the heating systems of three hospitals: a pair RDD of (hospital, kWh) in three partitions. filter keeps or drops each record where it lies, so output partition 1 is built from input partition 1 alone, on the same machine, in the same task. reduceByKey adds up the kWh of each hospital, so all the records of H1 must meet in one partition — and they lie in partitions 1 and 3. Spark cannot know in advance which hospitals a partition holds, so every input partition is cut by key into one piece per output partition, and every output partition fetches its piece from every input partition: a shuffle.

The slide of p. 1214 names the two cases. Narrow transformations, "1 to 1": filter(..), drop(..), coalesce(). Wide transformations, "(shuffles) 1 to N": distinct(), groupBy(..).sum(), repartition(n). Its pictures count from the input side: one input partition to one output, one input partition to N outputs. Its definitions count from the output side, and the narrow one gets it wrong.

The difference decides three things. Cost: a shuffle writes the pieces to the senders' local disks and pulls them over the network — the shuffle of MapReduce (lesson 30). Scheduling: an output partition after a shuffle needs pieces from all the input partitions, so every input task must finish first; the job is cut there, while narrow steps are chained inside one task. Recovery: a lost narrow partition is rebuilt from its own input partition, a lost partition after a shuffle from pieces of all of them.

The same operation can be either. A join is wide when Spark does not know where each key lives; when both inputs are already partitioned by the join key with the same partitioner, partition i of one meets partition i of the other and nothing moves: narrow, as the RDD paper of 2012 notes (and as PySpark 4.2.0 confirms).

Narrow: each input partition feeds at most one output partition, and nothing moves. Wide: an input partition is split among many outputs — a shuffle, and a new stage.

From RDDs to DataFrames

"With Spark 2.0, Spark is moving towards a DataFrame based syntax. But the way files are being distributed can still be thought of as RDDs, it is just the typed out syntax that is changing" (p. 1215). A DataFrame is "Immutable Data with named columns (built on RDDs)", with a "User-friendly API", "Uniform APIs across languages (Scala, Java, Python, R, and SQL)" and "Improved performance via optimizations (Tungsten and Catalyst)" (p. 1216) — the execution layer and the optimizer of lesson 40. Partitions, laziness, lineage, narrow and wide: everything above still holds.

The timeline of p. 1204 says what changed. An RDD (2011) is a "Distribute collection of JVM objects" with "Functional Operators (map, filter, etc.)": Spark runs your functions but cannot look inside them. A DataFrame (2013) is a "Distribute collection of Row objects" with "Expression-based operations and UDFs" and "Logical plans and optimizer": Spark can read col("age") > 30, optimize the plan and store the rows compactly. A Dataset (2015) is "Internally rows, externally JVM objects": typed and fast, "But slower than DF". The years are the slide's; the DataFrame API itself shipped with Spark 1.3, in 2015. Since Spark 2.0 they are one API (p. 1219): "DataFrame = Dataset[Row]" is the untyped alias, "Dataset[T]" the typed one, for Scala and Java. Python has only DataFrames — all the exam needs.

The code example of p. 1217 queries "different data sources" two ways: register DataFrames as temporary views and write a SQL string, or chain DataFrame methods. Both build the same plan. Corrected, they read:

# route 1: register views, then a SQL string
jsonDf = spark.read.json("s3n://…")
jsonDf.createOrReplaceTempView("json")
peopleDf.createOrReplaceTempView("people")
results = spark.sql("""SELECT p.id, p.name, j.country
                       FROM people p JOIN json j ON p.id = j.id
                       WHERE p.age > 30""")

# route 2: the DataFrame API, the same plan
results = (peopleDf.join(jsonDf, peopleDf.id == jsonDf.id)
                   .filter(peopleDf.age > 30)
                   .select(peopleDf.id, "name", "country"))

The paper of 30 Jan 2026 accepts only the second route: "Note that you are not allowed to use SQL syntax. Use PySpark instead." Lesson 39 trains it on every past SparkQL exercise.

Which is faster? Fig. 5 redraws the chart of p. 1218, "Time to aggregate 10 million integer pairs (in seconds)". RDD code depends on the language — Python about 9.5 s, Scala about 4.2 s — because Python lambdas run in separate Python processes, every record serialized to reach them. DataFrame code does not: about 2 s in SQL, R, Python and Scala alike, because the API only describes the computation and the optimized plan runs inside the JVM whatever language wrote it.

That is the fine print of "Available APIs" (p. 1226): "You can write in Java, Scala or Python", with an "interactive interpreter" for "Scala & Python only", and "Java & Scala are faster thanks to static typing" — true of RDD code, where Python pays for its separate processes; for DataFrames the chart shows no gap. (R and SQL, listed on p. 1216, are missing here.)

A DataFrame is an RDD of rows with named columns: the same partitions and laziness, plus a plan that Spark itself can read and optimize.

The life of a job — and of a lost partition

"Spark runs as a library in your program (1 instance per app)" (p. 1220): in the architecture of the slides you do not send queries to a Spark server, as you would to a database. Your program is the driver. Its one SparkContext runs tasks in local threads or obtains executors — processes on the worker machines — from a cluster manager, in "Mesos, YARN or standalone mode"; the tasks read storage themselves, "via Hadoop InputFormat API" (HBase, HDFS, S3). Fig. 6 draws who talks to whom (pp. 1206, 1220); the "Master" and "Slave" of p. 1207 are the driver and the executors.

What runs on the executors? The slide of p. 1224 shows the job of p. 1221 (log, then errors) at two levels. The dataset-level view shows objects: log is a HadoopRDD with path = hdfs://..., errors a FilteredRDD with func = _.contains(…) and shouldCache = true, the flag that cache() sets. The partition-level view shows the work: four partitions of log above four of errors, and one task per column — Task 1, Task 2, … — that reads its partition and filters it in one pass. A stage runs one task per partition.

The slide of p. 1225 follows an action through four components (Fig. 7). The transformations, rdd1.join(rdd2).groupBy(…).filter(…), have built a DAG of RDD objects. The action hands it to the DAGScheduler, which will "split graph into stages of tasks" and "submit each stage as ready": a stage ends wherever a wide dependency begins. Each stage goes, as a TaskSet of one task per partition, to the TaskScheduler, which will "launch tasks via cluster manager" and "retry failed or straggling tasks". On each worker, threads "execute tasks" and the block manager will "store and serve blocks": cached partitions and shuffle output. The words to keep: a job is everything one action triggers, a stage is a piece of it without shuffles, a task is one stage on one partition.

"RDDs track lineage information that can be used to efficiently recompute lost data" (p. 1223). When a worker dies, the partitions it held in memory are gone. The driver looks up their lineage — the HDFS file, filter and map of Fig. 4 — and schedules tasks on other workers that re-read the matching blocks, which HDFS still keeps on other DataNodes, and re-apply the functions, for the lost partitions only. Nothing is copied while all goes well, and recovery costs in proportion to what was lost. HDFS protects stored blocks by replication; Spark protects computed partitions by lineage. (Background, from the RDD paper: when a lineage grows very long, an RDD can be checkpointed to stable storage to cut it.)

Back to the puzzle of the opening. The four lines computed nothing: each added a node to the lineage, and cache() only set a flag. The first count ran the recipe and left each worker's partition of messages in its memory — hence half a second per search instead of a pass over the disk. The lost partition came back from the same recipe: one task re-read its block from a surviving HDFS replica and re-applied filter and map. Nobody had copied it.

An action becomes a job; the DAGScheduler cuts it into stages at the shuffles; each stage runs one task per partition; a lost partition is recomputed from its lineage.