SpiralTrain
Exercises › Block 2 · Exercise 5

Block 2 · Exercise 5

Window Functions

Open the notebook 17-window-functions-starter in your workspace and run its setup cell, which starts the Spark session and loads the flights as flights. Steps 2 to 5 have a challenge cell. In steps 1 and 6 the code is already in a cell : run it and answer the questions.

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 17-window-functions-solution, under every part on the exercise site, and at the back of the exercises PDF.

Step 1Flights

1a

Run the cell under Step 1. It prints the schema and the first five rows. Find dep_delay in both. Which values mean that a flight left early, and which mean that it was cancelled? Write the answer in a comment.

Hint 1

Look at the sign of the numbers, and at the rows that show null.

Check your output
root
 |-- flight_date: date (nullable = true)
 |-- carrier: string (nullable = true)
 |-- flight_number: integer (nullable = true)
 |-- tail_number: string (nullable = true)
 |-- origin: string (nullable = true)
 |-- dest: string (nullable = true)
 |-- dep_delay: double (nullable = true)
 ...

+-----------+-------+-------------+-----------+------+----+---------+...
|flight_date|carrier|flight_number|tail_number|origin|dest|dep_delay|...
+-----------+-------+-------------+-----------+------+----+---------+...
|        ...|    ...|          ...|        ...|   ...| ...|      ...|...
+-----------+-------+-------------+-----------+------+----+---------+...
only showing top 5 rows

The schema continues with the remaining columns, and the rows depend on the data set you loaded.

Show solutionHide solution
python
flights.printSchema()
flights.show(5)
# dep_delay is negative when a flight left early, and null when it was cancelled.
# Both matter in step 4.

Step 2Ranking

The starter defines a window that groups the rows by departure airport and orders each group from most to least delayed :

python
worst_first = Window.partitionBy("origin").orderBy(F.desc("dep_delay"))

2a

Assign to ranked the flights with three added columns over worst_first, row_number, rank and dense_rank, one for each ranking function. Then look at one busy airport, ATL : show origin, carrier, dep_delay and the three columns, twenty rows.

Hint 1

flights.withColumn("row_number", F.row_number().over(worst_first)) adds the first one. The other two follow the same pattern.

Hint 2

ranked.filter(F.col("origin") == "ATL").select(...) narrows it to one airport.

Check your output
+------+-------+---------+----------+----+----------+
|origin|carrier|dep_delay|row_number|rank|dense_rank|
+------+-------+---------+----------+----+----------+
|   ATL|    ...|      ...|         1|   1|         1|
|   ATL|    ...|      ...|         2| ...|       ...|
...
+------+-------+---------+----------+----+----------+
only showing top 20 rows

The rows are the most delayed departures at ATL, first. The carriers and delays depend on the data set you loaded.

Show solutionHide solution
python
ranked = (flights
    .withColumn("row_number", F.row_number().over(worst_first))
    .withColumn("rank", F.rank().over(worst_first))
    .withColumn("dense_rank", F.dense_rank().over(worst_first)))

ranked.filter(F.col("origin") == "ATL").select(
    "origin", "carrier", "dep_delay", "row_number", "rank", "dense_rank"
).show(20)

2b

Find two flights in the output that are delayed by the same number of minutes. Do the three columns agree on them? Which column leaves gaps in its numbering, and which does not? Write the answers in a comment.

Hint 1

Read the three columns on the row of the tie and on the row after it.

Show solutionHide solution
python
# Where two flights tie, row_number still gives them different numbers.
# rank and dense_rank give them the same number. rank leaves a gap after the tie,
# dense_rank does not.

Step 3Worst flight

3a

Use worst_first to keep one row per airport : the single most delayed departure. Assign it to worst_per_airport. Number the rows with row_number, keep number 1 and drop the helper column. Show the ten most delayed rows.

Hint 1

Add a column rn, filter on rn == 1, and drop rn again.

Check your output
+-----------+-------+-------------+...+------+----+---------+...
|flight_date|carrier|flight_number|...|origin|dest|dep_delay|...
+-----------+-------+-------------+...+------+----+---------+...
|        ...|    ...|          ...|...|   ...| ...|      ...|...
+-----------+-------+-------------+...+------+----+---------+...
only showing top 10 rows

One row per airport, with all the columns of flights, the highest dep_delay first.

Show solutionHide solution
python
worst_per_airport = (flights
    .withColumn("rn", F.row_number().over(worst_first))
    .filter(F.col("rn") == 1)
    .drop("rn"))

worst_per_airport.orderBy(F.desc("dep_delay")).show(10)

3b

Run 3a a second time. Do you get the same flights both times? Then make the result deterministic : assign to worst_first_stable a window with two more ordering columns, and repeat 3a with it as worst_per_airport_stable.

Hint 1

Where two flights at one airport share the worst delay, row_number has to pick one. Something in the row has to break the tie, such as the date and the carrier.

Hint 2

orderBy takes several columns : orderBy(F.desc("dep_delay"), "flight_date", "carrier").

Check your output
+-----------+-------+-------------+...+------+----+---------+...
|flight_date|carrier|flight_number|...|origin|dest|dep_delay|...
+-----------+-------+-------------+...+------+----+---------+...
|        ...|    ...|          ...|...|   ...| ...|      ...|...
+-----------+-------+-------------+...+------+----+---------+...
only showing top 10 rows

The same shape as 3a. Run it twice and the rows stay the same.

Show solutionHide solution
python
worst_first_stable = Window.partitionBy("origin").orderBy(
    F.desc("dep_delay"), "flight_date", "carrier"
)

worst_per_airport_stable = (flights
    .withColumn("rn", F.row_number().over(worst_first_stable))
    .filter(F.col("rn") == 1)
    .drop("rn"))

worst_per_airport_stable.orderBy(F.desc("dep_delay")).show(10)
# If two flights at one airport share the worst delay, row_number can pick either one
# on a rerun. Two more ordering columns make the choice fixed.

Step 4Airport average

4a

The starter defines by_airport = Window.partitionBy("origin"), a window with a partition and no ordering. Use it to attach the airport's average departure delay to every row, as a column airport_avg, and add a column vs_airport with the flight's own delay minus that average. Assign the result to compared and show ten rows of origin, carrier, dep_delay, airport_avg and vs_airport.

Hint 1

F.avg("dep_delay").over(by_airport) is the average per airport, on every row.

Hint 2

vs_airport needs the column airport_avg, so add that column first.

Check your output
+------+-------+---------+-----------+----------+
|origin|carrier|dep_delay|airport_avg|vs_airport|
+------+-------+---------+-----------+----------+
|   ...|    ...|      ...|        ...|       ...|
+------+-------+---------+-----------+----------+
only showing top 10 rows

Every row of one airport shows the same airport_avg.

Show solutionHide solution
python
compared = (flights
    .withColumn("airport_avg", F.avg("dep_delay").over(by_airport))
    .withColumn("vs_airport", F.col("dep_delay") - F.col("airport_avg")))

compared.select("origin", "carrier", "dep_delay", "airport_avg", "vs_airport").show(10)

4b

Assign to by_airport_ordered the same window with .orderBy("flight_date") added, and compute the average again as running_avg. Show origin, flight_date, dep_delay, airport_avg and running_avg, ten rows. The numbers change. Why? Write the answer in a comment.

Hint 1

The slide Window Frames says what a window covers when it has an ordering and when it has none.

Check your output
+------+-----------+---------+-----------+-----------+
|origin|flight_date|dep_delay|airport_avg|running_avg|
+------+-----------+---------+-----------+-----------+
|   ...|        ...|      ...|        ...|        ...|
+------+-----------+---------+-----------+-----------+
only showing top 10 rows
Show solutionHide solution
python
by_airport_ordered = Window.partitionBy("origin").orderBy("flight_date")

(compared
    .withColumn("running_avg", F.avg("dep_delay").over(by_airport_ordered))
    .select("origin", "flight_date", "dep_delay", "airport_avg", "running_avg")
    .show(10))
# An ordering changes the default frame from the whole partition to a running frame that
# ends at the current row, so running_avg is the average up to that day.

Step 5Previous flight

The starter defines a window over one route, ordered by date :

python
by_route = Window.partitionBy("carrier", "origin", "dest").orderBy("flight_date")

5a

Use F.lag("dep_delay", 1) over by_route to add prev_delay, the previous flight's delay on the same route, and add change, the difference between the two delays. Assign the result to trend. Show fifteen rows from ATL with carrier, origin, dest, flight_date, dep_delay, prev_delay and change. What are prev_delay and change on the first flight of every route?

Hint 1

change is dep_delay minus prev_delay, so add prev_delay first.

Check your output
+-------+------+----+-----------+---------+----------+------+
|carrier|origin|dest|flight_date|dep_delay|prev_delay|change|
+-------+------+----+-----------+---------+----------+------+
|    ...|   ATL| ...|        ...|      ...|       ...|   ...|
+-------+------+----+-----------+---------+----------+------+
only showing top 15 rows
Show solutionHide solution
python
trend = (flights
    .withColumn("prev_delay", F.lag("dep_delay", 1).over(by_route))
    .withColumn("change", F.col("dep_delay") - F.col("prev_delay")))

trend.filter(F.col("origin") == "ATL").select(
    "carrier", "origin", "dest", "flight_date", "dep_delay", "prev_delay", "change"
).show(15)
# The first flight on every route has no previous flight, so prev_delay and change
# are null there.

5b

Give lag a default value, F.lag("dep_delay", 1, 0), and build trend_default the same way. Show the same fifteen rows and check that the first flight of every route now has a number.

Hint 1

The third argument of lag is the value it gives when there is no previous row.

Check your output
+-------+------+----+-----------+---------+----------+------+
|carrier|origin|dest|flight_date|dep_delay|prev_delay|change|
+-------+------+----+-----------+---------+----------+------+
|    ...|   ATL| ...|        ...|      ...|       ...|   ...|
+-------+------+----+-----------+---------+----------+------+
only showing top 15 rows

The same columns as 5a, with no null in prev_delay or change on a route's first flight.

Show solutionHide solution
python
trend_default = (flights
    .withColumn("prev_delay", F.lag("dep_delay", 1, 0).over(by_route))
    .withColumn("change", F.col("dep_delay") - F.col("prev_delay")))

trend_default.filter(F.col("origin") == "ATL").select(
    "carrier", "origin", "dest", "flight_date", "dep_delay", "prev_delay", "change"
).show(15)

Step 6Plan

6a

Run the first cell under Step 6. It asks Spark for the plan of trend. Find the Window node and the Exchange below it.

Hint 1

Read the == Physical Plan == from the bottom up. An Exchange is a shuffle.

Check your output
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [...]
   +- Window [lag(...) windowspecdefinition(carrier, origin, dest, flight_date ...)]
      +- Sort [carrier ..., origin ..., dest ..., flight_date ...]
         +- Exchange hashpartitioning(carrier, origin, dest, ...)
            +- ...

The exact nodes, numbers and order of the lines vary with the Spark version and the data.

Show solutionHide solution
python
trend.explain()
# The Exchange is a shuffle : every row of one route has to arrive on the same executor
# before the window can be evaluated.

6b

Run the second cell under Step 6. It adds a window that partitions by the airport only and asks for the plan again. How many Exchange nodes are there now?

Hint 1

Look at the columns each Exchange partitions by. Does the route partitioning also put all the rows of one airport together?

Check your output
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Window [row_number() windowspecdefinition(origin, flight_date ...)]
   +- Sort [origin ..., flight_date ...]
      +- Exchange hashpartitioning(origin, ...)
         +- ...
            +- Window [lag(...) ...]
               ...

The exact nodes, numbers and order of the lines vary with the Spark version and the data.

Show solutionHide solution
python
by_origin_date = Window.partitionBy("origin").orderBy("flight_date")

(trend
    .withColumn("airport_rn", F.row_number().over(by_origin_date))
    .explain())
# The first plan has one Exchange, on the route. The second window partitions on origin
# alone, which the route partitioning does not satisfy, so this plan adds a second Exchange.
# Two windows partitioned the same way share one shuffle. Two windows partitioned
# differently cost one shuffle each.

Step 7Stop the session

7a

Click the session indicator and stop the session. A session holds its executors until the idle timeout, so a session left running can stop somebody else from starting theirs.

If time permits

  • Use ntile(4) to put every flight into a delay quartile within its airport, then count how many flights of each carrier land in the worst quartile.
  • Average the delay per airport and day, number the days with F.datediff, and build a seven-day rolling average with rangeBetween(-6, 0). Compare it to the seven-row rowsBetween(-6, 0) version. Where do they differ, and why?
  • Find, for every airport, the longest run of consecutive days on which the average departure delay rose. lag and a running sum will get you there.

Tried it yourself first?

The solution is a spoiler. Work through the hints first : a wrong attempt teaches more than a solution you only read.