Block 2 · Exercise 7
Open the notebook 19-performance-starter in your workspace and run the setup cell at the top. It starts the Spark session, which takes a few minutes ; read Step 1 while it starts. Steps 4, 6 and 7 have a cell with None where your code goes, Step 8 is an open task on a slow query, and the other steps run as they stand.
Every timing you take is your own. With the whole class working at once, a query runs slower than on the quiet capacity where the slides were measured. Compare your own runs with each other, and with nobody else's numbers.
Do one part at a time, 1a, 1b and so on, and compare with the expected output before you go on. The solution is in 19-performance-solution, under every part on the exercise site, and at the back of the exercises PDF.
The setup cell loads four tables :
flights from flights_large : the real flights of the previous modules, for more months, stored in one folder per year and monthairports : the OurAirports tablepings and receivers : generated tables for Step 4, joined on receiver_id. pings has one row per position report from an aircraft, receivers one row per ground receiverRun the cell of Step 1. It counts the flights per month. How many months are there, and roughly how many flights in each?
+----+-----+------+
|year|month| count|
+----+-----+------+
| ...| ... | ...|
+----+-----+------+One row per month, in order. The counts depend on the size of the table you have.
Run the cell of Step 2. It answers three questions with one query each. Write down how many flights, carriers and departure airports flights holds, and how many airports in airports have at least one departure.
+-------+--------+------------------+
|flights|carriers|departure_airports|
+-------+--------+------------------+
| ...| ...| ...|
+-------+--------+------------------+
airports with at least one departure : ...The last query of the cell shows the five airports with the most departures and their share of all flights. Which airport has the most departures, and what share of all flights is that? Write both down : it is the skew in this data, and Step 4 comes back to it.
+------+------+-----+
|origin| count|share|
+------+------+-----+
| ...| ...| ...|
+------+------+-----+Five rows, the busiest first.
A route is an origin and a dest together. The cells of Step 3 count the distinct routes flown on every day.
Run the first cell of Step 3. It shows the ten days with the most routes, then prints the formatted plan.
+-----------+------+
|flight_date|routes|
+-----------+------+
| ...| ...|
+-----------+------+
== Physical Plan ==
AdaptiveSparkPlan ...Ten rows, with routes falling from the top. The plan lists its nodes, and its shape depends on the Spark version.
Read the plan before you run anything else. How many Exchange nodes are there, and how many HashAggregate nodes? Why more than one HashAggregate? Count them in code and write the answer to the last question in a comment.
The slide Reading a Query Plan shows how the nodes are listed. The first block of the formatted output is the tree.
Exchange nodes : ...
HashAggregate nodes : ...tree = plan_text(routes_per_day).split(chr(10) + chr(10))[0]
print("Exchange nodes :", tree.count(" Exchange ("))
print("HashAggregate nodes :", tree.count(" HashAggregate ("))
# An aggregation runs twice : a partial one inside every partition before the shuffle,
# and a final one after it that combines the partial results. Counting distinct routes
# adds a step for the distinct pairs, so there are more than two.
# Every Exchange is a shuffle : the aggregation's and the sort's.Run the second cell of Step 3. It runs the query with collect() and prints the plan again. What does the first line say now? Look for AQEShuffleRead.
The slide Adaptive Query Execution says when AQE chooses its plan.
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
...
+- AQEShuffleRead ...AQEShuffleRead appears where AQE merged small partitions after a shuffle.
Would show() in place of collect() have given the final plan? Write why in a comment.
show() does not run the DataFrame's own plan again. What does it run?
# No. show() runs a new query of its own, so the DataFrame's plan stays the initial one,
# isFinalPlan=false. The SQL tab of the Spark UI shows the final plan either way.The real flights are not skewed enough to show a hot key. From 4c on, the join uses pings, a generated table in which receiver 1 hears nine pings in ten. It is made up on purpose : it is the clearest way to see what a hot key does. Every callsign in it starts with SIM.
Run the first cell of Step 4. It turns automatic broadcasting off, so Spark has to shuffle both sides, joins every flight to its departure airport and averages the departure delay per airport type.
+--------------+---------+
| type|avg_delay|
+--------------+---------+
| ...| ...|
+--------------+---------+One row per airport type.
Open the Spark UI and find the stage that performed the join. In its summary metrics, compare the maximum task duration with the median. How far apart are they? Which airport do you expect the slowest task held?
Look at the answer to 2b.
# Expected : the busiest airport of Step 2. Every row of one key hashes to the same partition,
# so the task that holds the hot key has the most rows and takes the longest.Run the second cell of Step 4. It switches AQE's skew handling off, so that only your own fix can help. It selects only the columns the join and its result need : a shuffle moves every column it is given. It joins pings to receivers on receiver_id and times the join with a noop write. Write down the time, and the maximum and the median task duration of the join stage in the Spark UI.
plain join : ... sThe time is your own.
Salt the join, for the hot receiver only. The key is an integer, so the salt goes in a second column and not on the end of the key. Fill in the cell with salted_pings and salted_receivers :
ping_cols, add a column salt : a random number from 0 to 7 on the rows of receiver 1, and 0 on every other rowreceiver_cols, replicate the row of receiver 1 eight times, once per salt value, and give every other receiver salt 0The slide Salting a Join shows the idea on a different table. F.when(...).otherwise(...) chooses per row.
F.rand(seed=7) gives a number between 0 and 1. Multiply and cast to int. F.explode of an array turns one row into as many rows as the array has items.
HOT = 1
SALTS = 8
salted_pings = ping_cols.withColumn(
"salt", F.when(F.col("receiver_id") == HOT, (F.rand(seed=7) * SALTS).cast("int")).otherwise(0))
salted_receivers = receiver_cols.withColumn(
"salt", F.explode(F.when(F.col("receiver_id") == HOT, F.array(*[F.lit(i) for i in range(SALTS)]))
.otherwise(F.array(F.lit(0)))))Join the two salted tables on both columns, ["receiver_id", "salt"], and drop the salt afterwards. Print the number of rows in receiver_cols and in salted_receivers. Time the salted join with a noop write, as in 4c, and check that it gives as many rows as the plain join.
.drop("salt") after the join keeps the result the same shape as the plain join.
receivers : ... rows ; salted : ... rows
salted join : ... s
same rows : TrueThe salted receivers hold seven more rows than the receivers. The time is your own.
salted_join = salted_pings.join(salted_receivers, ["receiver_id", "salt"]).drop("salt")
print("receivers :", receiver_cols.count(), "rows ; salted :", salted_receivers.count(), "rows")
start = time.time()
salted_join.write.format("noop").mode("overwrite").save()
print(f"salted join : {time.time() - start:.1f} s")
print("same rows :", salted_join.count() == ping_cols.join(receiver_cols, "receiver_id").count())Open the join stage of the salted join in the Spark UI. Did the maximum task duration fall? Did the total time fall, or did the extra rows eat the gain? Compare your times from 4c and 4e and write the answer in a comment.
# There is no number to copy : the answer is in your two times and your two maximum task durations.
# The salt splits receiver 1 over eight partitions, so the maximum task duration should fall.
# If the total time did not fall with it, the copies of receiver 1's row and the extra work
# on the ping side ate the gain.Run the next cell. It switches AQE's skew handling back on, joins pings to receivers again, and aggregates a checksum over the three selected columns so that the shuffle keeps all of them. It reads the final plan for SortMergeJoin(skew=true). If AQE did not split a partition, the cell lowers AQE's thresholds and tries again : you would not do that at work. Open the query in the SQL tab of the Spark UI as well.
The slide AQE Skew Join names the two settings that decide whether AQE splits a partition.
AQE split a partition : TrueIf it prints False, the cell lowers the thresholds and prints a second line. The plan that follows contains SortMergeJoin(skew=true) when a partition was split.
Run the cell that puts the settings back.
Run the cell of Step 5. It builds carriers, a table of eight carriers typed in by hand, and sets the broadcast threshold to Spark's own 10 MB, so the step behaves the same on every platform. It joins the flights to carriers twice and averages the arrival delay per carrier name : once as it is, once with broadcast(carriers). It prints the join each plan uses.
+------------+-------------+
|carrier_name|avg_arr_delay|
+------------+-------------+
| ...| ...|
+------------+-------------+
as it is [...]
broadcast() [...]One row per carrier. Each list names the join strategy in the plan, and the side it builds.
Call explain() on plain and on hinted. Which join does each plan use, and which side does it build, BuildLeft or BuildRight?
The slide Analyze Broadcast Join shows what a broadcast plan looks like.
== Physical Plan ==
...Join ...Two plans, one after the other.
plain.explain()
hinted.explain()
# hinted : BroadcastHashJoin with BuildRight, the right side is the carriers.
# plain : normally SortMergeJoin, see 5c.Why did Spark not broadcast an eight-row table on its own? Write the answer in a comment.
The slide Broadcast Join Threshold says what Spark compares with the threshold. Where does carriers come from, and what does Spark know about its size?
# A DataFrame built from a Python list carries no size estimate, so Spark cannot know it is small.
# A table read from parquet, such as airports, is broadcast without being asked.
# Spark estimates the other side too : on a small copy of the flights, the two columns the join
# needs can fall under the threshold, and the plan then broadcasts the flights, BuildLeft.Open both queries in the Spark UI. Which number disappears from the flights side of the join when you broadcast?
The slides Shuffle and Broadcasting say what moves across the network in each join.
# The shuffle write of the flights side. The sort-merge join shuffles the flights to meet the
# carriers ; the broadcast join leaves them where they are and sends the small table to them.For every departure airport, find the carrier with the second most departures and the carrier with the fewest.
Count the departures per origin and carrier into per_carrier, with the count named departures.
per_carrier = (flights.groupBy("origin", "carrier")
.agg(F.count("*").alias("departures")))Rank the carriers twice over the same partitioning, once from most departures and once from fewest, with dense_rank. Add the two ranks to per_carrier as rank_desc and rank_asc, in ranked. Fill in the cell with ranked = None.
You need two window specifications that both partition by origin and order in opposite directions. The slides Worst Delay per Airport and Ranking Functions of module 17 show the shape.
F.desc("departures") orders from most to fewest.
most_first = Window.partitionBy("origin").orderBy(F.desc("departures"))
least_first = Window.partitionBy("origin").orderBy("departures")
ranked = (per_carrier
.withColumn("rank_desc", F.dense_rank().over(most_first))
.withColumn("rank_asc", F.dense_rank().over(least_first)))Keep rank_desc == 2 for the second-busiest carrier and rank_asc == 1 for the least busy, and build answer with one row per airport and the columns origin, carriers, second_busiest and least_busy. Handle two edge cases :
Show the ten airports with the most carriers.
Count the carriers per airport from per_carrier. Join the three small tables back together with left joins on origin.
F.when(...).otherwise(...) handles both edge cases, one per column.
+------+--------+--------------+----------+
|origin|carriers|second_busiest|least_busy|
+------+--------+--------------+----------+
| ...| ...| ...| ...|
+------+--------+--------------+----------+Ten rows, carriers falling from the top. The values depend on the data.
second = ranked.filter(F.col("rank_desc") == 2).select("origin", F.col("carrier").alias("second_busiest"))
least = ranked.filter(F.col("rank_asc") == 1).select("origin", F.col("carrier").alias("least_busy"))
served = per_carrier.groupBy("origin").agg(F.count("*").alias("carriers"))
answer = (served.join(second, "origin", "left").join(least, "origin", "left")
.withColumn("second_busiest",
F.when(F.col("carriers") == 1, F.lit("only carrier")).otherwise(F.col("second_busiest")))
.withColumn("least_busy",
F.when(F.col("least_busy") == F.col("second_busiest"), F.lit(None)).otherwise(F.col("least_busy"))))
answer.orderBy(F.desc("carriers")).show(10)Look at the plan of ranked. How many Exchange nodes does it have, and how many Sort nodes? Would a window partitioned by carrier in place of origin add a shuffle? Write the answer to the second question in a comment.
The slide EnsureRequirements says when Spark adds an Exchange to a plan.
Exchange nodes : ... ; Sort nodes : ...tree = plan_text(ranked).split(chr(10) + chr(10))[0]
print("Exchange nodes :", tree.count(" Exchange ("), "; Sort nodes :", tree.count("Sort ("))
# Yes. A window needs the rows of its partition together. Both windows here partition by origin,
# so one Exchange serves both. A window partitioned by carrier needs the rows grouped another
# way, and that is a further Exchange.Add a column hashed_tail to every flight :
flight_number is even, apply MD5 to tail_number once for every capital N in it, feeding each hash into the nextflight_number is odd, apply SHA-256 to tail_number onceThen check whether any two flights share a hashed_tail.
Write the column as a Python UDF with hashlib. A tail number or a flight number can be missing, so the function has to return None for those. Register it as a UDF returning a string and add the column to the flights as with_udf. Fill in the cell with with_udf = None.
The slide UDF shows a UDF with F.udf and a return type. pyspark.sql.types.StringType is the return type here.
tail.count("N") gives the number of capital Ns. A for loop over range(...) applies the hash that many times. The hashes come from hashlib.md5(...).hexdigest() on encoded text.
import hashlib
from pyspark.sql.types import StringType
def hash_tail(tail, flight_number):
if tail is None or flight_number is None:
return None
if flight_number % 2 == 0:
hashed = tail
for _ in range(tail.count("N")):
hashed = hashlib.md5(hashed.encode()).hexdigest()
return hashed
return hashlib.sha256(tail.encode()).hexdigest()
hash_tail_udf = F.udf(hash_tail, StringType())
with_udf = flights.withColumn("hashed_tail", hash_tail_udf("tail_number", "flight_number"))Time with_udf with measure and a noop write, which runs the UDF on every row.
measure("label", lambda: noop(df)) times the call and prints the seconds, and under them read, shuffle write, spill and task times.
Python UDF ... s
read ... MB | shuffle write ... MB | spill ... MB memory, ... MB disk | ... tasks, slowest ... s, median ... sThe numbers are your own.
udf_run = measure("Python UDF", lambda: noop(with_udf))Write it without a UDF. F.sha2(col, 256) and F.md5(col) are built in ; the loop is the hard part. Count the Ns in tail_number with built-in functions first, and find the largest count in the data.
Take the length of the tail number, take the length of the same text with every N removed, and subtract. F.regexp_replace removes them.
most N in one tail number : ...without_n = F.regexp_replace("tail_number", "N", "")
n_count = F.length("tail_number") - F.length(without_n)
most_n = flights.agg(F.max(n_count)).first()[0]
print("most N in one tail number :", most_n)With the largest count known, write the loop as a chain of F.when clauses, and add the column to the flights as with_builtin. Fill in the cell with with_builtin = None. Missing values give None, as in 7a.
Build a list of columns : the tail number, then F.md5 of the item before it, most_n times. The clause for Ns counted k picks item k.
Wrap the even and the odd case in one outer F.when, the way 7a's function does with if.
chain = [F.col("tail_number")]
for _ in range(most_n):
chain.append(F.md5(chain[-1]))
even = F.when(n_count == 0, chain[0])
for k in range(1, most_n + 1):
even = even.when(n_count == k, chain[k])
hashed = (F.when(F.col("tail_number").isNull() | F.col("flight_number").isNull(), F.lit(None))
.when(F.col("flight_number") % 2 == 0, even)
.otherwise(F.sha2("tail_number", 256)))
with_builtin = flights.withColumn("hashed_tail", hashed)Time with_builtin the same way as 7b.
built-in functions ... s
read ... MB | shuffle write ... MB | ...The numbers are your own.
builtin_run = measure("built-in functions", lambda: noop(with_builtin))Do both versions give the same column? Compare them with exceptAll on the key columns flight_date, carrier, flight_number, origin and hashed_tail. Then count the hashes that more than one flight shares.
select(key) on both DataFrames, then exceptAll in one direction. An empty result means every row of the first is in the second.
rows differing : 0
hashes shared by more than one flight : ...key = ["flight_date", "carrier", "flight_number", "origin", "hashed_tail"]
print("rows differing :", with_udf.select(key).exceptAll(with_builtin.select(key)).count())
print("hashes shared by more than one flight :",
with_builtin.groupBy("hashed_tail").count().filter(F.col("count") > 1).count())Which version is faster, by your own times from 7b and 7e? Which plan shows BatchEvalPython? Check both plans with code.
The slide Without UDF says where the Python work shows up in a plan.
UDF BatchEvalPython in the plan : True
built-in BatchEvalPython in the plan : Falsefor name, df in (("UDF", with_udf), ("built-in", with_builtin)):
print(f"{name:9} BatchEvalPython in the plan : {'BatchEvalPython' in plan_text(df, 'simple')}")
# Faster : compare your two times. The UDF moves every row out of the JVM to Python and back.Would the built-in version still be correct if next month's data had a tail number with one more N ? Write the answer in a comment.
# No. The chain of F.when clauses has one clause for every count up to the largest count found
# in today's data. A tail number with more Ns matches none of them and gives null. The UDF
# loops as often as it needs to.The starter has a slow query. Use it, or write one of your own against the course data. Made up, not from work : never paste a query or data from work into this environment. It is a foreign tenant, and everything here is made up or public.
Run the cell with the slow query. It times the query twice with measure and prints its formatted plan. Write down the slower and the faster run.
slow query, first run ... s
read ... MB | shuffle write ... MB | ...
slow query, second run ... s
read ... MB | shuffle write ... MB | ...
== Physical Plan ==
...The times are your own, and the second run is often faster than the first.
Read the plan of slow. Mark every Exchange, the join strategy, the Python node, and what the scan filters on.
The slide What to Measure First lists what to look for, and the slide Partition Pruning says how a filter shows up in a scan.
BatchEvalPython : True
join, slow query : [...]text = plan_text(slow)
print("BatchEvalPython :", "BatchEvalPython" in text)
print("join, slow query :", join_used(slow))
# PartitionFilters lists what the scan prunes on. The filter on year(flight_date) is not among them.In the Spark UI, write down the four numbers from the slides for the slowest stage of the slow query : duration, shuffle read and write, maximum against median task time, and spill. The line under each measure result gives a total of the same kind for the whole query.
The slide What to Measure First names the four numbers.
Change one thing. Put the changed query in faster, time it with measure in the cell provided, and read the four numbers again. Write one line : what you changed, which number moved, and by how much. A change that moved nothing is worth a line too.
Which node in the plan holds most of the time ? Change that one.
A change to the filter, the Python function or the join each has a slide : Partition Pruning, Without UDF, Broadcasting.
1. filter on year, the partition column ... s
read ... MB | shuffle write ... MB | ...The read figure falls and the time follows it. The numbers are your own.
band = (F.when(F.col("dep_delay").isNull(), "cancelled")
.when(F.col("dep_delay") <= 0, "early or on time")
.when(F.col("dep_delay") <= 15, "up to 15")
.when(F.col("dep_delay") <= 60, "15 to 60")
.otherwise("over 60"))
pruned = (flights
.filter(F.col("year") == 2024)
.withColumn("band", delay_band_udf("dep_delay"))
.join(carriers, "carrier")
.groupBy("carrier_name", "band").count())
change_1 = measure("1. filter on year, the partition column", lambda: noop(pruned))
# The filter on year cuts the bytes read : year is a partition column, year(flight_date) is not.Repeat 8d until you run out of ideas, one change at a time, each with its own line.
2. built-in functions for the band ... s
3. broadcast the carriers ... s
same answer : True
PartitionFilters on year : True
BatchEvalPython : False
join, slow query : [...]
join, after the changes : ['BroadcastHashJoin', 'BuildRight']The times are your own. Your list of changes may differ from these three.
no_udf = (flights
.filter(F.col("year") == 2024)
.withColumn("band", band)
.join(carriers, "carrier")
.groupBy("carrier_name", "band").count())
broadcast_too = (flights
.filter(F.col("year") == 2024)
.withColumn("band", band)
.join(broadcast(carriers), "carrier")
.groupBy("carrier_name", "band").count())
change_2 = measure("2. built-in functions for the band", lambda: noop(no_udf))
change_3 = measure("3. broadcast the carriers", lambda: noop(broadcast_too))
print("same answer :", sorted(slow.collect()) == sorted(broadcast_too.collect()))
text = plan_text(broadcast_too)
print("PartitionFilters on year :", "PartitionFilters: [isnotnull(year" in text)
print("BatchEvalPython :", "BatchEvalPython" in text)
print("join, slow query :", join_used(slow))
print("join, after the changes :", join_used(broadcast_too))
# The built-in band removes the Python node. The hint makes the carriers the side that is copied :
# without it Spark shuffles both sides, or on a small copy of the flights may broadcast the flights.Run the cell that unpersists anything you cached and puts back every setting you changed.
settings back to the session's defaultsClick the session indicator and stop the session. A stopped session hands its executors back to the class.
receivers?flights, then repeat the counts from Step 2 twice. Compare the first and the second run with the uncached ones. Did the cache pay for itself at all?flights to your own Files/scratch, partitioned by carrier, once as it is and once after repartition("carrier"). Count the files each write produced. The solution notebook has cells for all three.