SpiralTrain
Exercises › Block 2 · Exercise 3

Block 2 · Exercise 3

DataFrames

Open the notebook 15-dataframes-starter in your workspace and run the setup cell, which starts the Spark session. Steps 2, 4 and 8 have a cell with a comment where your code goes. The other steps have a cell that already runs : run it, then answer the questions about what it shows.

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

Step 1Schema

1a

Run the cell of Step 1. It prints the schema of flights and of airports, and the number of columns of each. How many columns does each one have?

Check your output
root
 |-- flight_date: ... (nullable = true)
 |-- carrier: ...
 ...
root
 |-- iata_code: ...
 ...
flights columns  : <a whole number>
airports columns : <a whole number>

One line per column, with its type and whether it can be null.

Show solutionHide solution
python
flights.printSchema()
airports.printSchema()
print("flights columns  :", len(flights.columns))
print("airports columns :", len(airports.columns))

1b

Which column in flights matches which column in airports? Which columns could be null, and which of those would you expect to be null in practice?

Hint 1

Look for columns that hold the same kind of value, a three-letter code.

Hint 2

A flight that never left has no departure time.

Show solutionHide solution

origin and dest in the flights hold the same three-letter code as iata_code in the airports, which is the join key for the rest of the exercise.

python
print("join key :", "origin / dest against iata_code")

Step 2Select and filter

2a

Build very_late : the flights that left more than an hour late, holding only the date, the carrier, the origin, the destination and the delay. Print its row count and show the first five rows.

Hint 1

The slides Select Columns and Where Filter show the two calls. One select and one filter.

Hint 2

The delay is in minutes.

Check your output
rows : <a whole number>
+-----------+-------+------+----+---------+
|flight_date|carrier|origin|dest|dep_delay|
+-----------+-------+------+----+---------+
|        ...|    ...|   ...| ...|      ...|
...
+-----------+-------+------+----+---------+
only showing top 5 rows

Five rows, each with a dep_delay above 60.

Show solutionHide solution
python
very_late = (flights
    .select("flight_date", "carrier", "origin", "dest", "dep_delay")
    .filter(F.col("dep_delay") > 60))

print("rows :", very_late.count())
very_late.show(5)

2b

Run .explain() on very_late and find the ReadSchema line. How many columns does it list, and why is that fewer than the number of columns in flights?

Hint 1

Parquet stores each column separately. What does that let the reader skip?

Check your output
== Physical Plan ==
... FileScan parquet ... ReadSchema: struct<flight_date:...,carrier:...,origin:...,dest:...,dep_delay:...>

The ReadSchema line names only the five columns. The rest of the plan varies.

Show solutionHide solution
python
very_late.explain()

ReadSchema lists five columns. Parquet is columnar, so selecting five columns means the reader never touches the others. The columns not selected are never read off storage.

2c

Swap the order, filter before select, and explain again. Does ReadSchema change?

Show solutionHide solution

It does not change. Catalyst pushes the column list down to the reader wherever the select sits in the chain, so the order is a matter of readability.

Step 3Derived column

3a

Run the cell of Step 3. It adds delay_hours, the departure delay in hours rounded to two decimals, and delay_grade, which reads cancelled, on time, slight, late or very late. Read the five rows and the count per grade.

Hint 1

F.when(...).when(...).otherwise(...) is SQL's CASE WHEN. The branches are tried from top to bottom.

Check your output
+-------+---------+-----------+-----------+
|carrier|dep_delay|delay_hours|delay_grade|
+-------+---------+-----------+-----------+
|    ...|      ...|        ...|        ...|
...
+-------+---------+-----------+-----------+
only showing top 5 rows

+-----------+------+
|delay_grade| count|
+-----------+------+
|        ...|   ...|
...
+-----------+------+

Up to five grades, the largest first. The counts depend on the dataset size.

Show solutionHide solution
python
graded = (flights
    .withColumn("delay_hours", F.round(F.col("dep_delay") / 60, 2))
    .withColumn("delay_grade",
                F.when(F.col("dep_delay").isNull(), "cancelled")
                 .when(F.col("dep_delay") <= 0, "on time")
                 .when(F.col("dep_delay") <= 15, "slight")
                 .when(F.col("dep_delay") <= 60, "late")
                 .otherwise("very late")))

graded.select("carrier", "dep_delay", "delay_hours", "delay_grade").show(5)
graded.groupBy("delay_grade").count().orderBy(F.desc("count")).show()

3b

Move the isNull() branch from the front to the end of the when chain, and count the flights in each grade again. What happens to the cancelled flights?

Hint 1

Comparing null against a number does not give false. What does it give, and which branch does a row fall into then?

Show solutionHide solution

The isNull() branch has to come first. A cancelled flight has no dep_delay, and comparing a null against a number gives null rather than false, so without that branch the cancelled flights fall through to otherwise and get counted as very late. Nothing warns you about the wrong answer.

python
print("total :", graded.count(), "against", flights.count())

Step 4Group by carrier

4a

Build by_carrier : for every carrier, four numbers in one agg. They are the flight count, the average departure delay, the count of flights more than fifteen minutes late, and that count as a percentage of the flights. Order it by the percentage, highest first, and show it.

Hint 1

The slide GroupBy shows agg with several aggregate expressions. Count the late flights with F.sum over a F.when(...) that gives 1 or 0.

Hint 2

The percentage is a withColumn on the result of the agg, because it needs two of its columns.

Check your output
+-------+-------+---------+-----+--------+
|carrier|flights|avg_delay| late|late_pct|
+-------+-------+---------+-----+--------+
|    ...|    ...|      ...|  ...|     ...|
...
+-------+-------+---------+-----+--------+

One row per carrier, late_pct falling down the column. The carriers and numbers depend on the dataset size.

Show solutionHide solution
python
by_carrier = (flights
    .groupBy("carrier")
    .agg(F.count("*").alias("flights"),
         F.round(F.avg("dep_delay"), 2).alias("avg_delay"),
         F.sum(F.when(F.col("dep_delay") > 15, 1).otherwise(0)).alias("late"))
    .withColumn("late_pct", F.round(100 * F.col("late") / F.col("flights"), 2))
    .orderBy(F.desc("late_pct")))

by_carrier.show(truncate=False)

4b

Which carrier is worst on the percentage, and which is worst on the raw count of late flights? Are they the same carrier, and if not, why not?

Hint 1

Sort the result on late to see the raw count.

Show solutionHide solution

The count ranks the big carriers first, and the percentage ranks the unpunctual ones first. A carrier with more flights has more late flights, so the two lists are not the same.

4c

Add F.count("dep_delay").alias("departed") next to F.count("*").alias("flights") and compare the two columns. Why do they differ, and which one belongs in the denominator of your percentage?

Hint 1

count("*") counts rows. What does count on a column do with nulls?

Check your output
+-------+-------+--------+---------+-----+--------+
|carrier|flights|departed|avg_delay| late|late_pct|
+-------+-------+--------+---------+-----+--------+
|    ...|    ...|     ...|      ...|  ...|     ...|
...
+-------+-------+--------+---------+-----+--------+

departed is never larger than flights.

Show solutionHide solution
python
by_carrier = (flights
    .groupBy("carrier")
    .agg(F.count("*").alias("flights"),
         F.count("dep_delay").alias("departed"),
         F.round(F.avg("dep_delay"), 2).alias("avg_delay"),
         F.sum(F.when(F.col("dep_delay") > 15, 1).otherwise(0)).alias("late"))
    .withColumn("late_pct", F.round(100 * F.col("late") / F.col("flights"), 2))
    .orderBy(F.desc("late_pct")))

by_carrier.show(truncate=False)

count("*") counts every row and count("dep_delay") skips the nulls, so the two are over different populations. avg skips the nulls too. Keep both counts when interpreting the average.

Step 5HAVING

5a

Run the cell of Step 5. It keeps only the carriers with more than a hundred thousand flights. Does the ranking change once the small carriers are gone?

Hint 1

There is no having() method. A filter after a groupBy is a HAVING, and a filter before it is a WHERE. The slide Filter on Aggregate Data shows it.

Check your output
+-------+-------+--------+
|carrier|flights|late_pct|
+-------+-------+--------+
|    ...|    ...|     ...|
...
+-------+-------+--------+

Only carriers with more than 100000 flights, late_pct falling down the column.

Show solutionHide solution
python
by_carrier.where(F.col("flights") > 100000) \
    .select("carrier", "flights", "late_pct").show(truncate=False)

The big carriers and the unpunctual ones are not the same list, which is the point of ranking on the percentage rather than the count.

Step 6Join airports

6a

Run the cell of Step 6. It joins the airports on the origin and shows the average delay per departure city, for the cities with more than ten thousand flights.

Hint 1

The slide Inner Join shows a join on two different column names.

Check your output
+------------+-------+---------+
|municipality|flights|avg_delay|
+------------+-------+---------+
|         ...|    ...|      ...|
...
+------------+-------+---------+
only showing top 10 rows

Up to ten cities, each with more than 10000 flights, avg_delay falling down the column.

Show solutionHide solution
python
by_city = (flights
    .join(airports, flights.origin == airports.iata_code)
    .groupBy("municipality")
    .agg(F.count("*").alias("flights"),
         F.round(F.avg("dep_delay"), 2).alias("avg_delay"))
    .filter(F.col("flights") > 10000)
    .orderBy(F.desc("avg_delay")))

by_city.show(10, truncate=False)

6b

Open the SQL tab of the Spark UI and look at the plan. Does it say BroadcastHashJoin or SortMergeJoin? Which side did Spark broadcast, and why that one? Print the plan in the notebook as well.

Hint 1

by_city.explain() prints the plan as text. Look for the join node, and for a BroadcastExchange under it.

Check your output
== Physical Plan ==
AdaptiveSparkPlan ...
+- Sort ...
   +- Exchange ...
      +- Filter ...
         +- HashAggregate ...
            ...
               +- BroadcastHashJoin ...

The shape is a scan, a join, two aggregates with an Exchange between them, then the filter and the sort. Names and numbers vary.

Show solutionHide solution
python
by_city.explain()

The airports table is small, so Spark usually broadcasts it rather than shuffling both sides. The plan says which it chose.

Step 7Row count

7a

Run the cell of Step 7. It prints three numbers : the flight count, the count after the inner join, and the count after the same join written as a left join. Are all three the same?

Hint 1

The slides Inner Join and Left Outer Join say which rows each keeps.

Check your output
flights      : <a whole number>
inner joined : <a whole number>
left joined  : <a whole number>

The left join keeps every flight, so its count equals the flight count.

Show solutionHide solution
python
joined = flights.join(airports, flights.origin == airports.iata_code)

print("flights      :", flights.count())
print("inner joined :", joined.count())
print("left joined  :", flights.join(
    airports, flights.origin == airports.iata_code, "left").count())

7b

If the inner join lost or gained rows, is that because a key was missing or because a key was duplicated? Check whether iata_code is unique in the airports table.

Hint 1

Compare the number of distinct values in iata_code with the number of rows.

Check your output
keys unique   : True
Show solutionHide solution
python
print("keys unique   :",
      airports.select("iata_code").distinct().count() == airports.count())

An inner join can only lose rows or multiply them. Unique keys on the right rule out multiplication, which leaves a missing key. The next step looks for it.

Step 8Anti join

8a

Build missing : the distinct origin codes in the flights that have no row in the airports table. Print how many there are, and show them.

Hint 1

The slide Left Semi Join and Left Anti Join shows the join type that keeps the rows without a match.

Hint 2

Take the distinct origins first, then join them against airports on iata_code.

Check your output
origins missing from the airports table : <a whole number>
+------+
|origin|
+------+
|   ...|
...
+------+

One row per missing code. The table is empty when the airports table covers every origin.

Show solutionHide solution
python
missing = (flights.select("origin").distinct()
    .join(airports, F.col("origin") == airports.iata_code, "leftanti"))

print("origins missing from the airports table :", missing.count())
missing.show()

8b

Build unused : the airports in the table that never appear as an origin. Print how many there are.

Hint 1

It is the same join, the other way round : the airports on the left, the distinct origins on the right.

Check your output
distinct origins                        : <a whole number>
airports in the table                   : <a whole number>
airports never used as an origin        : <a whole number>

The third number is most of the second.

Show solutionHide solution
python
unused = (airports.select("iata_code", "name")
    .join(flights.select("origin").distinct(),
          airports.iata_code == F.col("origin"), "leftanti"))

print("distinct origins                        :",
      flights.select("origin").distinct().count())
print("airports in the table                   :", airports.count())
print("airports never used as an origin        :", unused.count())

Most of the table is unused, because the airports file is global and the flights are domestic US.

8c

Compare the rows that the missing codes of 8a account for with the rows the inner join lost in 7a. Do they match?

Hint 1

Count the flights whose origin is one of the codes in missing.

Show solutionHide solution

An anti join finds out that a reference table is incomplete, before a report built on an inner join under-reports without an error. If missing is empty, the inner join lost nothing, and the three counts of 7a agree.

Step 9Struct column

The starter builds routes : flights with a struct column route, made of the origin as from and the destination as to.

9a

Show flight_date, carrier and route for three rows, without truncating. How does a struct print?

Hint 1

The slide Nested Struct DataFrame shows how a struct prints.

Check your output
+-----------+-------+--------------+
|flight_date|carrier|route         |
+-----------+-------+--------------+
|...        |...    |{<from>, <to>}|
...
+-----------+-------+--------------+
only showing top 3 rows

The struct prints as one value in braces.

Show solutionHide solution
python
routes = flights.withColumn(
    "route", F.struct(F.col("origin").alias("from"), F.col("dest").alias("to")))

routes.select("flight_date", "carrier", "route").show(3, truncate=False)

9b

Select route.from alone, then route.*. What is the difference? Print the schema of flight_date, route.* and dep_delay selected together.

Hint 1

A dot goes into a struct. A star takes out every field.

Check your output
root
 |-- flight_date: ...
 |-- from: string (nullable = true)
 |-- to: string (nullable = true)
 |-- dep_delay: ...

The struct is gone, and its two fields sit next to the other columns.

Show solutionHide solution
python
routes.select("route.from").show(3)
routes.select("route.*").show(3)
routes.select("flight_date", "route.*", "dep_delay").printSchema()

route.from gives one column. route.* gives one column per field of the struct, here from and to. select("route.*") flattens the struct.

9c

Try routes.select("flight_date", "route").toPandas(). What lands in the route cell?

Hint 1

Look at the type of one value in the route column.

Show solutionHide solution
python
pdf = routes.select("flight_date", "route").limit(3).toPandas()
print(pdf)

The struct does not arrive as two columns. Flatten a struct with select("route.*") before converting to pandas or writing to CSV, since neither handles a struct column.

Step 10Stop session

10a

Click the session indicator and stop the session.

Show solutionHide solution

Nothing to run. A session holds its executors until the idle timeout whether you are running anything or not, and with the whole class on one capacity a forgotten session is why somebody else cannot start theirs.

If time permits

  • Answer Step 4 again for the routes, grouping on origin and dest rather than on the carrier. There are thousands of groups instead of a dozen. Watch the shuffle in the Stages tab and compare it against the carrier version.
  • Write the Step 6 result to parquet under SCRATCH, read it back, and check that the schema survived the trip.
  • Find the single worst departure delay in the data, and the airport and carrier it belongs to. Then find the worst delay that was not a cancellation.

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.