The sales job from the previous lesson worked, but every new question cost one more job: sales by producer and market was one, the ranking per market another, joining it with the catalogue a third, and between each of them the complete output went to HDFS and came back. Apache Spark was born at Berkeley (2009, Matei Zaharia) out of precisely that frustration: iterative machine learning algorithms and interactive queries took ten or a hundred times longer in MapReduce than the CPU justified, because the time went on writing to and reading from disk between phases. Spark keeps the essentials of the model (data in partitions, independent and re-runnable tasks, a shuffle to group by key) but expresses the whole computation as a DAG of operators that a scheduler turns into stages and that runs while keeping the intermediate data in memory. On that foundation it built a collections API (RDDs), then a tables API with an optimiser (DataFrames and Spark SQL), and on top of those, libraries for machine learning (MLlib), graphs (GraphX) and streams (Structured Streaming, which we will see in 05-04). Today it is the reference batch engine. In this lesson analytics rewrites the daily sales computation as services/analytics/daily_sales.py, with DataFrames and with RDDs for comparison, launches it locally and on a cluster from docker-compose.yml, reads the execution plan, and trains some recommendations with ALS on the clicks.

1. Why Spark: an in-memory DAG versus phases on disk
2. Architecture: driver, cluster manager, executors, tasks and stages
3. RDDs: immutable collections, lazy transformations and lineage
4. DataFrames and Spark SQL: Catalyst and columnar formats
5. Narrow and wide operations, shuffle and stages
6. Optimisations: cache, broadcast join, partitioning and skew
7. MapReduce versus Spark
8. MLlib: recommendations with ALS on the clicks
9. Hands-on: services/analytics/daily_sales.py
10. Common Mistakes and Tips
11. Exercises
12. Conclusion

In MapReduce (05-02), a chain of three groupings is three jobs and six trips through disk: each job writes its output to HDFS (replicated three times) and the next one reads it. For an iterative algorithm that goes over the same data a hundred times, that is a hundred full reads from disk of a dataset that has not changed. Spark changes two things:

• The whole computation is a single program, a DAG of operators (read, filter, explode, group, join, write) that the engine knows in full before running anything. With that global view it can chain into a single pass the operators that do not need to redistribute data, and it only materialises intermediate data at the points where there is a shuffle. This is the dataflow pattern from 05-01.
• Intermediate data lives in the memory of the executors, and can be explicitly cached to be reused across several passes. A hundred iterations over a cached dataset read the disk once.

The price of not writing to disk is fault tolerance: if a node dies, its intermediate data in memory disappears. Spark's solution, and its original idea, is lineage (section 3): instead of saving the intermediate data, it saves the recipe for recomputing it from the input, and recomputes only the lost partitions. It is the deterministic re-execution of 05-01, applied to fragments of a computation rather than to whole tasks.

• Driver. The process that runs the main program (daily_sales.py). It holds the SparkSession (formerly SparkContext), builds the DAG, divides it into jobs, stages and tasks, sends them to the executors and collects results. It is the master of 05-02, one per application: if the driver dies, the application dies (on YARN in cluster mode, the driver is the ApplicationMaster and can be retried).
• Cluster manager. It shares out resources among applications. Spark ships with its own (standalone, one master and workers: the one we will use in docker-compose.yml), and integrates with YARN (05-02: the driver requests containers from the ResourceManager) and with Kubernetes (07-05: each executor is a pod). The driver does not care which one it is; the API is the same.
• Executor. A JVM process on a worker node, with N cores and M GB assigned, that lives for the whole application. It runs tasks in threads (one task per core at a time) and keeps cached partitions and shuffle data in memory. Unlike MapReduce, no JVM is started per task: the JVM is started once and runs thousands of tasks, which eliminates the fixed per-task cost.
• Task. The unit of work: running a chain of operators over one partition. A stage with 200 partitions is 200 tasks.
• Stage. A set of tasks that can run without a shuffle. The DAG is cut into stages at every wide operation (section 5); within a stage, the operators are chained (pipelining) and a row goes through all of them without touching disk.
• Job. Everything triggered by an action (section 3): write, collect, count. A program may have several jobs; each job has one or more stages.

With PySpark, the driver is a Python process that talks to a JVM (via Py4J), and on the executors the Python code of RDD functions runs in auxiliary Python processes with round-trip serialisation; with DataFrames, by contrast, most operations are translated into JVM code and Python is not involved per row. That is the main reason the DataFrame API is the recommended one.

The RDD (Resilient Distributed Dataset) is Spark's original abstraction: a collection of elements partitioned across the executors, immutable (it is not modified: another RDD is created from it) and resilient through lineage. You operate on it with two kinds of methods:

Laziness is what makes optimisation possible: when the program says filter and then map and then count, Spark does not run three passes; on reaching count it knows the whole chain and runs it in a single pass per partition. And lineage is the DAG of transformations that led to each RDD: if a partition of the sales RDD is lost because its executor died, Spark looks at the lineage (sales = lines.reduceByKey(...), lines = events.flatMap(...), events = sc.textFile(...)) and recomputes only that partition from the corresponding HDFS block. Nothing intermediate has to be saved to disk; the cost is recomputing, which is affordable as long as the lineage is short (for long, iterative lineages there is checkpoint(), which does materialise to HDFS and cuts the lineage).

The sales computation with RDDs, which will serve as a comparison in the hands-on part:

Up to this point not a single byte has been read: textFile, map, filter, flatMap and reduceByKey are transformations. sales.collect() or sales.saveAsTextFile(...) would launch the job. Two details that set a Spark user apart: reduceByKey does the reduction locally in each partition before the shuffle (the combiner of 05-02, automatically), whereas groupByKey followed by a sum moves all the values over the network; and lambda functions travel serialised to the executors, so they cannot capture non-serialisable objects (a database connection, for example).

An RDD is a collection of objects that are opaque to Spark: it does not know that ln["producer"] is a column, so it cannot optimise anything beyond chaining. A DataFrame is a table with a schema (named, typed columns), distributed in partitions like an RDD, on which you operate with a relational API (select, filter, groupBy, join, agg) or directly with SQL (spark.sql("SELECT ...")). Both produce the same logical plan, which goes through Catalyst, the optimiser:

1. Analysis: resolving column names and types against the schema.
2. Logical optimisation: rules such as pushing filters down towards the read (predicate pushdown: filtering type = 'order.created' while reading, not afterwards), reading only the columns needed (column pruning), simplifying expressions, reordering joins.
3. Physical planning: choosing algorithms: a join is done by broadcast hash if one side is small, by sort-merge if not; an aggregation in two phases (partial in each partition, final after the shuffle).
4. Code generation: Tungsten compiles the plan into specialised Java bytecode (whole-stage codegen) and manages memory off the JVM heap in a compact binary format. This is what makes PySpark with DataFrames as fast as Scala: Python only describes the plan.

Since Spark 3, Adaptive Query Execution (AQE) re-optimises the plan during execution with real statistics: it reduces the number of partitions after a small shuffle, turns a sort-merge join into a broadcast if it discovers that one side fits, and splits skewed partitions (section 6).

DataFrames go hand in hand with columnar formats. A Parquet file stores the data by columns instead of by rows: all the producer values together, all the amount values together, compressed with encodings suited to each type (dictionary for repeated strings, run-length for consecutive values) and with statistics (minimum, maximum, nulls) per row group. A query that only needs producer and amount reads those two columns and skips the rest of the file, and a filter day = '2026-09-14' skips entire row groups whose range does not contain it. Compared with JSONL, which forces you to read and parse every byte, Parquet cuts both the bytes read and the size on disk by an order of magnitude. That is why the lake of 04-02 starts out as JSONL (what Kafka produces) and the first step of the analytics converts it to Parquet, partitioned by day.

The division of the DAG into stages depends on a single criterion: what dependency each output partition has on the input ones.

The sales DAG has exactly one shuffle, the groupBy, and therefore two stages. Everything else is chained: a JSON row is filtered, exploded and partially aggregated in the same task without anybody writing it out. The join with the catalogue, which in MapReduce would be a third job, is narrow thanks to the broadcast (section 6). In the Spark web UI (http://localhost:4040 during execution) each job appears with its stages, each stage with its tasks, and for each stage the shuffle write and shuffle read bytes: these are the counters of 05-02, and the diagnostic criterion is the same.

Spark's shuffle also writes to local disk (the shuffle files, served by the executors or by an external shuffle service), so "in memory" does not mean "no disk": it means that between narrow operations nothing is written, and that cached data is served from memory. With 200 shuffle partitions by default (spark.sql.shuffle.partitions), a small job produces 200 tiny tasks in the second stage; AQE merges them, but on old versions or with AQE disabled it is worth lowering the number.

cache() and persist(). They mark a DataFrame or RDD so that, the first time it is computed, its partitions are kept (in memory by default; persist(StorageLevel.MEMORY_AND_DISK) allows spilling to disk, DISK_ONLY, or serialised to save memory). It pays off when the same intermediate result is used in several actions: the lines DataFrame in the hands-on part feeds the sales by producer, a ranking and a quality check; without the cache, every action reads and parses the JSON from HDFS again. unpersist() releases it. Caching something that is used once is pure cost.

Broadcast join. Joining the sales (millions of rows, spread out) with the catalogue (a few dozen rows) should not move the sales. With F.broadcast(catalog), Spark sends a copy of the catalogue to every executor and the join is resolved locally, with no shuffle of the large side: it is the map-side join of 05-02, automatically. Spark does it on its own if it estimates that the small side takes up less than spark.sql.autoBroadcastJoinThreshold (10 MB by default); forcing it with broadcast() is worthwhile when the estimate fails (a CSV with no statistics). With two large sides, the plan is a sort-merge join: a shuffle of both by the key and a sorted merge.

repartition(n) and coalesce(n). The number of partitions governs the parallelism. repartition(n) does a full shuffle to obtain n balanced partitions (or repartition("day") to place each day in its own partition, useful before a partitioned write); coalesce(n) reduces the number without a shuffle, merging local partitions, and is the cheap way of not writing 200 Parquet files of 3 KB. As a guide: partitions of 100–200 MB in memory and between 2 and 4 tasks per available core.

Skew with salting. Artisan Cheese Week is back: when grouping by producer, the montblanc-dairy partition receives half the rows. With DataFrames, the partial aggregation per partition (which Catalyst always inserts) eases the case of sums, just as the combiner did; skew hurts when the operation cannot be pre-reduced (a sort-merge join, collect_list, windows). The technique from 05-01, now with columns:

And since Spark 3, AQE with spark.sql.adaptive.skewJoin.enabled=true detects skewed join partitions (by size relative to the median) and splits them automatically, with no manual salt. For skewed aggregations with no possible pre-reduction, the salt is still needed.

Spark's advantage on the sales job of 05-02 shows in the numbers: the same computation that took 52 s in MapReduce on the test cluster takes 6 s in Spark on YARN with the same resources, and chaining the ranking and the join with the catalogue adds not jobs but a stage. The trade-off is memory: a badly sized executor (too little memory for the shuffle or the cache) fails with an OutOfMemoryError where MapReduce simply wrote to disk.

The second case in Kilometre Zero's analytics, "recommended products for Anna", is an iterative problem: exactly the kind that was handled badly in MapReduce. MLlib comes with ready-made distributed algorithms, and for recommendation the classic is ALS (Alternating Least Squares), a factorisation of the customers × products matrix that learns one factor vector per customer and another per product from interactions (purchases, clicks) and predicts the interest of each unobserved pair. With clicks there is no "rating", only implicit signals (viewed, added to basket, purchased), and ALS has a mode for that.

The clicks in the lake, /km0/clicks/2026-09-14/hour=13/web-01.jsonl (04-02), look like this:

Each iteration of fit is a job with several stages over the same data, which MLlib caches internally; with 10 iterations in MapReduce it would be 20 jobs and 20 reads from HDFS. The result (anna → crianza-wine, lucy → aged-cheese...) is loaded by the pipeline of 05-05 into the catalog database for the website to display. What MLlib does not decide is the quality: choosing rank, regParam and the weights of the actions requires evaluation (RegressionEvaluator or ranking metrics over a held-out set), and that belongs in a machine learning course, not this one.

To the docker-compose.yml of 04-02 (HDFS) and 05-02 (YARN) we add a Spark standalone cluster, which is lighter than YARN for development. The driver will run on our machine or in the master's container:

The two workers offer 4 cores in total, and HDFS is reachable as hdfs://namenode:8020 from the same Compose network. The catalogue is a small CSV, services/analytics/catalog.csv, which in production would come from a daily export of km0_catalog:

Key points of the program:

• Explicit schema. Without it, spark.read.json does a full pass just to infer types. With it, the read is one pass and the types are the ones we want (timestamp_ms as LongType, not double).
• explode turns the lines array into rows, which is what the mapper's for ln in ev["data"]["lines"] loop did.
• A single shuffle. The groupBy is the only wide operation. filter, select, explode and the broadcast join are chained in the same stage as the read. Catalyst inserts a partial aggregation before the shuffle (HashAggregate with partial_sum), so what crosses the network is partials by (day, market, producer) per partition: the combiner, without writing it.
• cache() on lines because there are two actions (check.first() and the write); without it, the JSON would be read and parsed twice.
• partitionBy("day") with partitionOverwriteMode=dynamic. The output is one directory per day (day=2026-09-14/), and overwrite in dynamic mode replaces only the days this job writes, leaving the rest untouched. Running the job for 14 September twice produces exactly the same output: it is the per-partition idempotency that the pipeline of 05-05 needs for reprocessing and backfilling.
• coalesce(1) before writing, because the aggregate is about 12 rows per day and we do not want 8 Parquet files of 2 KB. With large aggregates you would remove or adjust it.

Locally, with four threads as simulated executors (local[4]), over the file generated in 05-01:

The plan is read from the bottom up, and every line confirms a decision from the previous sections: FileScan json with PushedFilters (the filter pushed down to the read), Generate explode, InMemoryRelation (the cache), the HashAggregate with partial_sum before the Exchange hashpartitioning(..., 8) (partial aggregation, then the only shuffle with 8 partitions) and the final HashAggregate, and BroadcastHashJoin with a BroadcastExchange of the CSV (the catalogue travels, the sales do not). If SortMergeJoin with two Exchange nodes appeared instead of BroadcastHashJoin, we would know that the broadcast was not applied and that we are paying for one shuffle too many.

Against the Compose standalone cluster, with the file in HDFS:

With --master yarn and the Hadoop configuration in HADOOP_CONF_DIR, the same file would be launched on the YARN of 05-02, with the driver as the ApplicationMaster (--deploy-mode cluster), without changing a line of code. The master's UI (localhost:8080) shows the workers and the applications; the driver's (localhost:4040, while it is running) shows the jobs, stages and tasks, with their shuffle bytes.

sales_rdd.py contains the function from section 3. Launched with --rdd, it produces the same figures, and two differences show up in the UI: the read stage is slower (every line goes through a Python process with json.loads, instead of the JVM's JSON parser) and there is no plan to read: Spark runs the lambdas as they are, without pushing down filters or pruning columns, because it does not know what they do. On the 130 MB file the difference is 9 s versus 4 s in local[4]; on terabytes, hours. The RDD is still the right tool when the data has no schema or the logic does not fit into column expressions, and it is what lies underneath every DataFrame; but for analytics over events with a schema, DataFrames is the API.

• collect() on a large DataFrame. It brings every row to the driver, which has a few GB of memory: OutOfMemoryError on the driver. To take a look, show() or take(n); to save, write.
• Forgetting that transformations are lazy. A badly written filter does not fail when you write it, but at the first action, with a stack trace that points to the write. And a print inside a lambda does not show up on the driver: it runs on the executors (check their logs).
• groupByKey + sum on RDDs. It moves all the values over the network. reduceByKey or aggregateByKey combine locally. With DataFrames, groupBy().agg() already does so.
• Row-by-row Python UDFs. A udf in PySpark serialises every row out to a Python process and back; it defeats Catalyst and Tungsten. Almost everything can be expressed with F.*; if not, use pandas_udf (vectorised in batches with Arrow).
• Caching without using it, or never releasing. cache() on something used once is cost with no benefit; caching lots of things without unpersist() overflows the executors' memory and causes silent evictions and recomputations.
• 200 shuffle partitions for 10 MB. The default value of spark.sql.shuffle.partitions is meant for large clusters. Adjust it or trust AQE; and coalesce before writing so as not to produce hundreds of one-kilobyte files that will choke the NameNode (04-02).
• Trusting the automatic broadcast with a CSV. Without statistics, Spark may fail to estimate the size and do a sort-merge join. Use an explicit F.broadcast() and check it in explain().
• Inferring the JSON schema in production. It is an extra pass and a schema that changes with the data (one day without version and the column disappears). Use an explicit schema, versioned along with the event contract (02-05).
• Writing with overwrite without partitionOverwriteMode=dynamic. It deletes the whole output directory, including the days that were not being recomputed. It is the most expensive mistake made with a backfill pipeline.

Extend sales_dataframe so that, in addition to the sales, it produces for each (day, market) the top-selling producer and its share of the market total (as a percentage), using window functions (pyspark.sql.Window). How many shuffles does your solution add? Compare it with the two jobs of exercise 1 in 05-02.

This is the plan of a modified version of the program, written by a colleague. Identify four performance problems from the plan and say how to fix each one.

The events from 8 to 14 September 2026 (Grape Harvest Week) had an error in the price of crianza-wine that orders has corrected by regenerating the orders.jsonl files for those seven days in HDFS. Write the invocation (or invocations) of daily_sales.py that recomputes exactly those seven days without touching the rest, and explain which combination of the program's options guarantees that (a) no old data for those days is left behind, (b) the other days are not deleted, and (c) running it twice gives the same result. What would you change if the input were a glob orders-*.jsonl with several files per day?

Exercise 1.

The two windows share the partitioning (day, market), so Spark adds one shuffle (Exchange hashpartitioning(day, market)) and a Sort within each partition for row_number; with AQE and the same keys, it sometimes reuses the exchange. Total: two shuffles in the program (the groupBy and the window), in a single job with three stages, versus MapReduce's two jobs with their four trips through HDFS. And the join with the catalogue still costs no shuffle because sales already came with it resolved.

Exercise 2.

1. SortMergeJoin with two Exchange nodes by producer instead of BroadcastHashJoin: the three-row catalogue is causing a shuffle of the sales and a sort. Fix: F.broadcast(catalog).
2. BatchEvalPython [compute_amount]: a row-by-row Python UDF, which also prevents partial_sum from being computed in the JVM without going out to Python. Fix: F.col("ln.quantity") * F.col("ln.price").
3. Exchange ... 200: 200 shuffle partitions for an aggregate of a few dozen rows; 200 tiny tasks per stage. Fix: spark.sql.shuffle.partitions set to 8 (or AQE with coalescing enabled).
4. FileScan json [event_id, type, version, timestamp_ms, source, data] with no PushedFilters and no pruning: every column is read, and probably without an explicit schema (inference, an extra pass). Fix: a declared schema and a select of the columns needed right after the read; the type filter should appear under PushedFilters (it does when the column is compared with a literal and the schema is known).

A fifth detail: there is no InMemoryRelation, so if the program runs more than one action, it will re-read the JSON each time.

Exercise 3.

A single invocation with a glob for the seven days, or seven invocations (one per day, which is what Airflow will do in 05-05 with the backfill):

(a) mode("overwrite") replaces the contents of each day=2026-09-NN/ partition the job writes, atomically per partition (a write to a temporary location and a commit). (b) partitionOverwriteMode=dynamic limits the deletion to the partitions present in the job's output; without it, overwrite would empty the whole of daily_sales/, including August and the rest of September. (c) The computation is deterministic (same input, same aggregate) and the write replaces the complete partition: two runs leave the same bytes (apart from internal file names), without duplicating or accumulating. It is important that day is derived from timestamp_ms (event time) and not from the directory name: if an event from the 14th arrived late and orders had left it in the file for the 15th, the job for the 15th would write it to day=2026-09-14/, overwriting the partition for the 14th with only that event. To avoid this there are two options: filter in the job by the range of days being processed (F.col("day").between(...)) and discard or log those out of range, or make the input and output partitions match by construction. With a glob orders-*.jsonl per day nothing changes in the program (spark.read.json accepts globs and directories); it is only worth checking that none of the files is still being written, which is exactly what the sensor of 05-05 will watch for with _SUCCESS or with a closing file.

Spark takes the MapReduce model (partitions, re-runnable tasks, a shuffle by key) and strips out what made it slow: the entire computation is a DAG of operators that the driver knows in full, cuts into stages only where there is a wide dependency, chains everything else in the same task, and keeps the intermediate data in the memory of executors that live for the whole application. Fault tolerance comes from lineage, which recomputes the lost partition instead of having written it to disk. On top of RDDs, immutable and lazy, DataFrames add a schema and an optimiser, Catalyst, which pushes down filters, prunes columns, inserts partial aggregations and chooses broadcast or sort-merge for each join; Tungsten compiles the plan, and Parquet makes reading two columns from a year of events cost what those two columns take up. We have learnt to read an explain() to verify that the plan does what we think, to use cache when there are several actions, broadcast for the catalogue, coalesce before writing, salt for Montblanc Dairy, and partitionBy with dynamic overwrite so that the daily sales job is idempotent per day. services/analytics/daily_sales.py now computes in six seconds, and in a single program, what in 05-02 was three jobs and a minute; and ALS in MLlib has turned the clicks in the lake into recommendations without the hundred iterations meaning a hundred reads.

Everything we have done starts from a bounded dataset: the file for 14 September, already closed, or seven days of clicks. But orders.events never closes: stock.updated events arrive at a rate of hundreds per second, and the couriers' positions at 2.4 million a day, and neither the delivery dashboard nor the low-stock alert can wait for the nightly batch. The next lesson deals with stream processing: what changes when the dataset never ends, how "the last five minutes" is defined when events arrive out of order (watermarks), and how Flink and Spark Structured Streaming run the same DAG as this lesson over a stream that never stops.