Programming Spark: key-value operations and PySpark DataFrames, exam-style

RDDs by hand: create, transform, act

The exercise session opens with one sentence: Apache Spark "is a multi-language, unified analytics engine for big data processing, with built-in modules for executing data engineering, data science, and machine learning tasks on single-node machines or clusters" (p. 1712). An engine, not a store: the lecture's "Importers for most data formats and from most database technology" (p. 1227) reach Cassandra, Elasticsearch, MongoDB, Neo4j, Redis and CSV, JSON, Avro or Parquet files, and the session draws Spark at the hub of streaming, ETL, machine-learning, analysis, graph and data-lake tools (p. 1716; p. 1717 sorts the tools by kind). Spark reads the data where it lives, computes, and writes the result back.

Every program starts from an entry point. The lecture calls it the SparkContext: "Main entry point to Spark functionality", "Available in shell as variable sc", and "In standalone programs, you'd make your own" (p. 1228). Since Spark 2.0 you build a SparkSession — the first line of every session notebook, and a safe first line for an exam answer — and reach the context through it:

# the entry point (E08, p. 1726); sc is the SparkContext of p. 1228
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("SMBUD").getOrCreate()
sc = spark.sparkContext

The session spells the name RDD out (p. 1720): Resilient, "Fault-tolerant"; Distributed, "Computed across multiple nodes"; Dataset, a "Collection of partitioned data". Its next slide lists four characteristics (p. 1721): RDDs "Are immutable", "Work with lazy evaluation", "Are divided into partitions" and "Are fault-tolerant, which is achieved through lineage graphs" — lesson 38 shows how each works.

An RDD comes from one of three sources (p. 1229): a Python collection, cut into partitions by parallelize; a text file read by textFile, one element per line; or any existing Hadoop InputFormat. The path chooses the file system — "file.txt" is local, "hdfs://namenode:9000/path/file" is HDFS, a glob such as "directory/*.txt" reads many files — so the same program runs on a laptop and on a cluster.

# two RDDs: a Python list, and the lines of a file
nums = sc.parallelize([1, 2, 3])
lines = sc.textFile("hamlet.txt")

Transformations build, actions run

Three transformations do most of the work (p. 1230). map passes each element through a function: one in, one out. filter keeps the elements for which a predicate is true. flatMap maps each element to zero or more elements and flattens the result: range(x) is the sequence 0, 1, …, x − 1, so 1, 2 and 3 become [0], [0, 1] and [0, 1, 2] — six elements in all.

# the three basic transformations of p. 1230, with what they hold
squares = nums.map(lambda x: x*x)               # [1, 4, 9]
even = squares.filter(lambda x: x % 2 == 0)     # [4]
nums.flatMap(lambda x: range(x))                # [0, 0, 1, 0, 1, 2]

None of these lines computes anything. A transformation returns a new RDD and records how to build it; an action makes Spark run the recorded steps and either return a value to the driver program or write the result out (p. 1231):

# the basic actions of p. 1231: only these make Spark work
nums.collect()                          # [1, 2, 3], to the driver
nums.take(2)                            # [1, 2]
nums.count()                            # 3
nums.reduce(lambda x, y: x + y)         # 6
nums.saveAsTextFile("hdfs://file.txt")  # part files, not one file

collect is fine for three numbers and fatal for a terabyte: it copies the whole dataset into the memory of the driver, while take(n) fetches only n elements. saveAsTextFile writes a directory of part files, one per partition. The exercise session sorts the operations into two lists (p. 1722):

Pairs: reduceByKey, the word count and the shuffle

"Spark's "distributed reduce" transformations operate on RDDs of key-value pairs" (p. 1232). Nothing has to be declared: in Python a pair is a 2-tuple, pair[0] is the key and pair[1] the value (Scala writes pair._1 and pair._2, Java builds a Tuple2), and an RDD whose elements are pairs gets the …ByKey operations. The slide's example has three pets (p. 1233):

# one value per key: merge the values with the function
pets = sc.parallelize([("cat", 1), ("dog", 1), ("cat", 2)])
pets.reduceByKey(lambda x, y: x + y)    # [('cat', 3), ('dog', 1)]

# all the values of a key; the pairs sorted by key
pets.groupByKey()     # [('cat', [1, 2]), ('dog', [1])]
pets.sortByKey()      # [('cat', 1), ('cat', 2), ('dog', 1)]

reduceByKey merges all the values of one key with the function; groupByKey only collects them, one list per key; sortByKey orders the pairs by key. Under the code the slide adds the sentence that carries the "why": "reduceByKey also automatically implements combiners on the map side". A combiner is the trick of the MapReduce word count of lesson 30: each machine first merges the values of the same key that it holds, and only the partial results travel. groupByKey cannot do that, because it must deliver every single value. Same answer, different traffic — Fig. 1.

The word count of lesson 30 took a page of Java. Here it is three chained calls on the lines of hamlet.txt (p. 1234):

# the word count of p. 1234: three transformations, then an action
lines = sc.textFile("hamlet.txt")
counts = lines.flatMap(lambda line: line.split(" ")) \
              .map(lambda word: (word, 1)) \
              .reduceByKey(lambda x, y: x + y)
counts.collect()
# [('be', 2), ('not', 1), ('or', 1), ('to', 2)] on the slide's lines

Read it as a flow of data. flatMap turns each line into its words ("to be or" gives to, be, or); map turns each word into the pair (word, 1); reduceByKey is the shuffle plus the reduce: each partition's partial count of a word travels to the machine responsible for that word — a hash of the key decides which — and is added up there. On the slide the arrows from the two lines cross: (be, 2) and (not, 1) end on one machine, (or, 1) and (to, 2) on the other. Only collect() starts the job.

Two RDDs of pairs can be joined on their keys (p. 1235). join returns one pair per matching combination, the two values side by side; cogroup returns one pair per key, with the list of values from each side:

# join: one output pair per matching (left value, right value)
visits = sc.parallelize([("index.html", "1.2.3.4"),
                         ("about.html", "3.4.5.6"),
                         ("index.html", "1.3.3.1")])
pageNames = sc.parallelize([("index.html", "Home"),
                            ("about.html", "About")])
visits.join(pageNames)
# ('index.html', ('1.2.3.4', 'Home'))
# ('index.html', ('1.3.3.1', 'Home'))
# ('about.html', ('3.4.5.6', 'About'))

# cogroup: one output pair per key, with both lists of values
visits.cogroup(pageNames)
# ('index.html', (['1.2.3.4', '1.3.3.1'], ['Home']))
# ('about.html', (['3.4.5.6'], ['About']))

Two visits of index.html and one name give two join rows but one cogroup row. Finally, "All the pair RDD operations take an optional second parameter for number of tasks" (p. 1236): the result is cut into that many partitions, each worked by one reduce task. Too few and the cluster idles; too many and the scheduling and the tiny output files cost more than they save.

# 5 reduce tasks: the result has 5 partitions
words.reduceByKey(lambda x, y: x + y, 5)

Under the hood: stages, and why iterations fly

When an action runs, the DAG scheduler turns the recorded transformations into work for the cluster. The slide lists what it does: "General task graphs", "Automatically pipelines functions", "Data locality aware", "Partitioning aware to avoid shuffles" (p. 1237). Its figure, redrawn as Fig. 2, shows all four.

The graph is cut into stages at every shuffle, because a shuffle is a barrier: no partition after it can start before all the partitions before it are done. Inside a stage the narrow operations — each input partition feeds exactly one output partition (lesson 38) — are pipelined: map and filter on C run as one task per partition, in one pass, and D is never stored. Tasks go where their partition already is (data locality, as with HDFS blocks). B already came out of a groupBy, partitioned by key, so the join takes it as it is, and B is cached, so Stage 1 is skipped.

The exercise session says the same in two views (pp. 1723–1724). In the logical view "An RDD is seen as a single dataset", the transformations "form a lineage graph", and "everything is just a plan". In the physical view "An RDD is split into partitions", "Each partition lives on a worker node", is "processed independently and in parallel", and "Fault tolerance is achieved by recomputing only the lost partitions". The API also has more operators than map and reduce (p. 1238): outer joins, cogroup, cross (the Cartesian product, called cartesian in PySpark), zip, sample, fold, and partitionBy, the tool behind "partitioning aware".

Why the bars are so short

Hadoop was born when "RAM was extremely expensive, so Hadoop was designed to rely heavily on hard disk drives (HDDs) for storage and processing" (p. 1713): every MapReduce job reads its input from HDFS and writes its output back. Spark came when RAM had become affordable, and "was designed to leverage RAM for in-memory computation, drastically reducing the time required for iterative tasks such as machine learning and interactive queries" (p. 1715). Between the two, the session's slide on HDFS (p. 1714) has one sentence to correct before you quote it.

Fig. 3 puts the lecture's benchmark numbers on one scale.

K-means and logistic regression — both classic methods of statistics — pass over the same data once per iteration and update a tiny model, centroids or a weight vector. With the data cached in memory, an iteration costs one pass in RAM; as a chain of Hadoop jobs, it costs a read from HDFS, a job start and a write back. PageRank gains less because every iteration still shuffles the ranks across the network. Doubling the machines more than halved Hadoop's time per iteration (171 to 80 s) but cut Spark's only from 23 to 14 s: the shorter an iteration, the larger the share of fixed costs that more machines do not split.

The last benchmark slide is the strongest: "On-Disk Sort Record: Time to sort 100TB" — Hadoop, 2100 machines, 72 minutes; Spark, 207 machines, 23 minutes; "Also sorted 1PB in 4 hours" (p. 1241, from the Daytona GraySort benchmark of sortbenchmark.org). 100 TB does not fit in memory, so RAM is not the whole story: a third of the time on a tenth of the machines is about 30 times more data sorted per machine per minute. That is the engine itself — stages, pipelining, a better shuffle.

The DataFrame: an RDD that knows its columns

The lecture introduces the DataFrame with a dictionary entry (p. 1246):

Picture a pandas DataFrame whose rows are spread over many machines, that cannot be changed in place, and whose operations wait for an action. The next slide gives the reasons (p. 1247): it is "A distributed collection of rows with the same schema (RDDs suffer from type erasure)"; it "Can be constructed from external data sources or RDDs into essentially an RDD of Row objects"; it "Supports relational operators (e.g. where, groupby) as well as Spark operations"; and it is "Evaluated lazily → unmaterialized logical plan". Type erasure is a Java argument — at run time an RDD of Person objects cannot tell Spark which fields a Person has; in Python, an RDD of tuples has no column names and no types either. A DataFrame carries both, so Spark can check a query, store the rows compactly and optimise the plan before running it.

The session's slide puts it in one line: "DataFrames, on the other hand, are RDDs with an added schema that defines column names and data types". RDDs stay "Best for unstructured data or custom transformations", and "DataFrames are preferred today because they run much faster, use less memory, are easier to write and maintain, and integrate with SQL and other paradigms/systems" (p. 1725). The speed comes from what the schema allows: Spark sees the whole query and optimises it with Catalyst (lesson 40), while a Python lambda in RDD code is a black box it can only call.

The data model is nested (p. 1248): primitive SQL types — boolean, integer, double, decimal, string, date, timestamp — and complex types, "structs, arrays, maps, and unions", with "First class support for complex data types". That is why an exam table can hold a list in a cell: the Hospital row of the paper of 30 Jan 2026 has Heating_Systems = [HS_1, HS2, ...], a column of type array.

A query can be written in two ways (p. 1249): "Relational operations (select, where, join, groupBy) via a DSL", whose "Operators take expression objects" and "build up an abstract syntax tree (AST), which is then optimized by Catalyst" — or, "Alternatively, register as temp SQL table and perform traditional SQL query strings". The slide's example is Scala, where === is column equality; in PySpark it reads:

# female employees per department (p. 1249)
from pyspark.sql import functions as F

(employees
    .join(dept, employees["deptId"] == dept["id"])
    .where(employees["gender"] == "female")
    .groupBy(dept["id"], dept["name"])
    .agg(F.count(employees["name"])))

The count names its table, employees["name"]: if both tables have a column called name, a bare count("name") is ambiguous and Spark refuses it.

What the papers call "SparkQL" is this API, and the paper of 30 Jan 2026 adds to each query: "Note that you are not allowed to use SQL syntax. Use PySpark instead." So no spark.sql("SELECT …") — and, the safe reading, no SQL expression string inside filter either. The session shows one query in Python, SQL and Java (p. 1726); its PySpark box is the skeleton of every answer:

# the PySpark box of p. 1726: session, data, query, action
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("PythonExample").getOrCreate()
df = spark.read.csv("hdfs://data/sales.csv", header=True, inferSchema=True)
df.groupBy("region").sum("amount").show()

The SQL box next to it is not quite the same query: it names the total total_sales and sorts it, largest first, while sum("amount") makes a column called sum(amount), in no particular order. The PySpark twin of the SQL box uses agg with alias — the form the exam answers need:

# the SQL box of p. 1726, without SQL
df.groupBy("region").agg(F.sum("amount").alias("total_sales")) \
  .orderBy(F.col("total_sales").desc()).show()

The Java box has a slip of its own: it reads the CSV without inferSchema, so amount stays a string and sum("amount") stops with "amount" is not a numeric column (checked on PySpark 4.2). header=True takes the column names from the first line; inferSchema=True reads the file once more to guess the types.

Above the engine sit the libraries — Spark SQL, which "manages Datasets and DataFrames", Spark Streaming, MLlib and GraphX — and below it a cluster manager that hands out the machines: YARN, Mesos, the Standalone Scheduler or Kubernetes (pp. 1727–1728; Spark 4.0 has since dropped Mesos, so the current documentation lists the other three). The engine serves batch data, "collecting data over a period of time and then processing it all at once", for high throughput, and real-time streams, processed "continuously as it arrives, with low latency": Spark Streaming (the older API) and Structured Streaming (the modern one) group the incoming events into micro-batches and run each with the batch engine (p. 1719).

The use-case slide (p. 1718) adds SQL analytics, exploratory data analysis "also on petabyte-scale datasets", and machine learning developed "on a single computer" and then run, the exact same code, "on large, fault-tolerant clusters with thousands of machines". The session notebooks are the single-computer end: one process on your laptop, .master("local").

The word count, once more, as a DataFrame. split turns each line into an array of words, explode makes one row per element, and groupBy("word").count() does what reduceByKey did:

# the word count with the DataFrame API: no lambdas, no pairs
lines_df = spark.read.text("hamlet.txt")      # one column, "value"
counts_df = (lines_df
    .select(F.explode(F.split(F.col("value"), " ")).alias("word"))
    .groupBy("word")
    .count())
counts_df.show()

Its physical plan, printed by counts_df.explain() on PySpark 4.2, has the same three moves as Fig. 1's right panel: a partial count inside each partition, an exchange of rows by a hash of the word, a final count. The widget runs both versions side by side.

The session's menu, one operation at a time

The practice of the exercise session is not in the slide PDF: it is in four notebooks published next to it on WeBeeP (E08 - Spark.zip) — data upload, filtering, grouping and joining, all on a pizza menu. The assistants' advice for every practical topic applies word for word: "focus on understanding how the query shapes the data. Think on how the dataset would change after each operation in the query is performed." The notebooks build the menu with an explicit schema, StructField(name, type, nullable):

# the menu of notebooks 1–3, built with an explicit schema
from pyspark.sql.functions import col, explode, array_contains
from pyspark.sql.functions import sum, avg, count, max, countDistinct
from pyspark.sql.types import (StructType, StructField, StringType,
                               FloatType, ArrayType)

schema = StructType([
    StructField("Pizza Name", StringType(), True),
    StructField("Price", FloatType(), True),
    StructField("Ingredients", ArrayType(StringType()), True)])
df_data = [("Margherita", 5.95, ["Tomato Sauce", "Mozzarella Cheese", "Basil"]),
           ("Calzone", 7.95, ["Tomato Sauce", "Mozzarella Cheese", "Prosciutto Cotto"]),
           ("Diavola", 5.95, ["Tomato Sauce", "Mozzarella Cheese", "Spicy Salame"]),
           ("Prosciutto", 7.95, ["Tomato Sauce", "Mozzarella Cheese", "Prosciutto Cotto"]),
           ("Speck & Brie", 7.95, ["Tomato Sauce", "Mozzarella Cheese", "Speck", "Brie"]),
           ("Tonno & Cipolle", 7.95, ["Tomato Sauce", "Mozzarella Cheese", "Tuna", "Onions"]),
           ("Fries", 3.95, ["Potatoes"])]
df = spark.createDataFrame(data=df_data, schema=schema)

With a list of column names instead of a schema — spark.createDataFrame(df_data, ["Pizza Name", "Price", "Ingredients"]) — Spark infers the types: string, double, array of strings. Either way the rows are those of Table 2. A DataFrame cannot be changed, so to "add one element" — the 1-point question of the 2026 paper — you build a one-row DataFrame with the same columns and union it, which matches the columns by position and returns a new DataFrame:

# adding a row: union returns a new DataFrame
new_row = spark.createDataFrame(
    [("Bufalina", 8.5, ["Tomato Sauce", "Buffalo Mozzarella"])], schema)
bigger_df = df.union(new_row)     # 8 rows; df still has 7

Filter: conditions are columns

A condition for filter, or for its synonym where, is an expression on columns: df.Price, df["Pizza Name"] or col("Pizza Name") — a name with a space needs one of the last two. Conditions are combined with & (and), | (or) and ~ (not), each comparison in its own parentheses (notebook 2):

# Price above 7.9 and the name Calzone: one row, Calzone
df.filter((df.Price > 7.9) & (col("Pizza Name") == "Calzone")).show()

The notebook's "N.B. Parenthesis are essential!" has a reason. Python binds & more tightly than > and ==, so without parentheses it first computes 7.9 & col("Pizza Name"), then asks whether a column is true — and the query never reaches Spark:

# the same filter without parentheses: Python stops before Spark runs
df.filter(df.Price > 7.9 & col("Pizza Name") == "Calzone").show()
# PySparkValueError: [CANNOT_CONVERT_COLUMN_INTO_BOOL] Cannot convert
# column into bool: please use '&' for 'and', '|' for 'or',
# '~' for 'not' when building DataFrame boolean expressions.

Python's and gives the same error. Text has its own methods: contains("&"), startswith("To"), endswith("one"), like("%on%") with SQL's wildcards, rlike with a regular expression; isin([...]) tests a list of values, and array_contains(df.Ingredients, "Basil") looks inside a list column.

Explode, group, aggregate

explode turns a list column into rows: one row per element, every other column copied (notebook 1). The seven pizzas have 3 + 3 + 3 + 3 + 4 + 4 + 1 = 21 ingredients, so 21 rows come out; without an alias the new column is called col. It is MongoDB's $unwind of lesson 15.

# one row per (pizza, ingredient): 21 rows
exploded_df = df.select(col("Pizza Name"), df.Price,
                        explode(df.Ingredients).alias("Ingredient"))

select keeps only the columns you name. withColumn adds or replaces one column and keeps all the others: df.withColumn("Ingredient", explode(df.Ingredients)) gives the same 21 rows with the Ingredients list still attached.

groupBy followed by count() or agg(...) collapses the rows to one per key (notebook 3). agg takes any number of aggregate functions, each renamed with alias: count, sum, avg, max, countDistinct for different values, and collect_list or collect_set, which gather a group's values back into a list, with or without duplicates — the reverse of explode. SQL's HAVING is a filter after the agg, on the alias:

# at least four ingredients: Speck & Brie 4, Tonno & Cipolle 4
exploded_df.groupBy("Pizza Name") \
    .agg(count("Ingredient").alias("Number of Ingredients")) \
    .filter(col("Number of Ingredients") >= 4) \
    .show()

Now the puzzle of the hook. The same notebook also computes sum("Price").alias("Sum Price") per pizza on exploded_df. Calzone's price was copied into each of its three ingredient rows, so the sum is three times the price: the code asks for "the price, once per ingredient, added up". The digits come from the schema: 7.95 stored as a float is 7.949999809265137, and three of those added as doubles make 23.84999942779541. After an explode, sum and count what each new row stands for — here the ingredients, in the exam the heating systems — never a value of the parent row, which the explode repeated; take that with max, or aggregate before exploding.

Join

Notebook 4 replaces the list with a foreign key — a pizza table (Pizza Name, Price, Ingredients_ID), which the notebook calls df and this lesson pizza_df — and joins it with ingredient_df (Ingredients_ID, Ingredients). Two pizzas point to IG_7 and IG_8, which the other table lacks, and IG_5 and IG_6 are never used, so every join type gives a different count (Table 3).

# the inner join of notebook 4: 4 rows, both key columns kept
pizza_df.join(ingredient_df,
    pizza_df.Ingredients_ID == ingredient_df.Ingredients_ID,
    "inner").show()

A join on an expression keeps both key columns, so a later select("Ingredients_ID") stops with [AMBIGUOUS_REFERENCE] Reference `Ingredients_ID` is ambiguous. Join on the name instead — on="Ingredients_ID" — and Spark keeps one key column, placed first; or rename a column before the join. The notebook calls leftsemi a "left inner join": it is not one — it returns each matching pizza once, with no columns of the other table, like SQL's WHERE … IN (…).

The exam exercise, end to end

The paper of 30 Jan 2026 (Exercise 4 - Spark, 5 PT) starts from the ER diagram of a hospital heating management system and draws two tables: Hospital (Hospital_ID, Name, Heating_Systems), with rows such as HOSP_1 | ... | [HS_1, HS2, ...], and Heating System (Heat_ID, H_KW_Cons, Heat_Power). "Consider only the attributes defined." Its first question:

"Write a Python code to create a dataframe for the first table and to add one element in the dataframe. (1PT)"

# (a) the Hospital table, plus one hospital: union and re-assign
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("Hospitals").getOrCreate()

hospital_cols = ["Hospital_ID", "Name", "Heating_Systems"]
hospital_rows = [("HOSP_1", "Milano Care Center", ["HS_1", "HS_2"]),
                 ("HOSP_2", "Como General", ["HS_3"])]
hospital_df = spark.createDataFrame(hospital_rows, hospital_cols)
new_row = spark.createDataFrame(
    [("HOSP_3", "Lecco Care Clinic", ["HS_4", "HS_5"])], hospital_cols)
hospital_df = hospital_df.union(new_row)

The rows are ours, as are those of the heating table (Table 4): the paper prints only "...". Spark infers string, string and array of strings. Write every value of a column the same way — 120.0 in every row, not 120 in one and 40.5 in another — or createDataFrame stops with [CANNOT_MERGE_TYPE].

The third question: "Write a SparkQL query to compute and return the total Heat_Power for each hospital, returning only the names of the hospitals whose total is greater than 100. (2 PT) Note that you are not allowed to use SQL syntax. Use PySpark instead." Translate it phrase by phrase, and look at the table after every step:

1. The list cannot be joined as it is: a join compares one value with one value. Explode it — one row per (hospital, heating system):

   .select("Hospital_ID", "Name",
           F.explode("Heating_Systems").alias("Heat_ID"))

   Hospital_ID	Name	Heat_ID
   HOSP_1	Milano Care Center	HS_1
   HOSP_1	Milano Care Center	HS_2
   HOSP_2	Como General	HS_3
   HOSP_3	Lecco Care Clinic	HS_4
   HOSP_3	Lecco Care Clinic	HS_5

2. Bring in Heat_Power with a join on the column name — Spark keeps one Heat_ID, placed first:

   .join(heating_df, "Heat_ID")

   Heat_ID	Hospital_ID	Name	H_KW_Cons	Heat_Power
   HS_1	HOSP_1	Milano Care Center	120.0	40.0
   HS_2	HOSP_1	Milano Care Center	120.0	70.0
   HS_3	HOSP_2	Como General	90.0	30.0
   HS_4	HOSP_3	Lecco Care Clinic	300.0	80.0
   HS_5	HOSP_3	Lecco Care Clinic	150.0	50.0

3. "the total Heat_Power for each hospital": group by the key and carry the name — two hospitals could share a name:

   .groupBy("Hospital_ID", "Name")
   .agg(F.sum("Heat_Power").alias("Total_Heat_Power"))

   Hospital_ID	Name	Total_Heat_Power
   HOSP_1	Milano Care Center	110.0
   HOSP_2	Como General	30.0
   HOSP_3	Lecco Care Clinic	130.0

4. "whose total is greater than 100": a HAVING, so a filter after the agg, on the alias:

   .filter(F.col("Total_Heat_Power") > 100)

   Hospital_ID	Name	Total_Heat_Power
   HOSP_1	Milano Care Center	110.0
   HOSP_3	Lecco Care Clinic	130.0

5. "returning only the names": a last select. The answer is two names.

   .select("Name")

   Name
   Milano Care Center
   Lecco Care Clinic

Put together, with the chain in parentheses so that it can span lines:

# (c) total Heat_Power per hospital; names of those above 100
from pyspark.sql import functions as F

totals = (hospital_df
    .select("Hospital_ID", "Name",
            F.explode("Heating_Systems").alias("Heat_ID"))
    .join(heating_df, "Heat_ID")
    .groupBy("Hospital_ID", "Name")
    .agg(F.sum("Heat_Power").alias("Total_Heat_Power"))
    .filter(F.col("Total_Heat_Power") > 100)
    .select("Name"))
totals.show()

Move the filter before the groupBy — .filter(F.col("Heat_Power") > 100) — and the answer is empty: no single heating system reaches 100; only the totals do. A filter before the grouping removes rows, like WHERE; after it, it removes groups, like HAVING. Table 5 collects the phrases of the three papers and their methods.

The same recipe — explode, join, filter, group, filter the totals, select — solves the other past papers. Try them in the exam questions below before opening the model answers, and test your code in the playground: its tables hold our test rows for each paper.