Analyzing airline data with Spark SQL

In [1]:
import findspark
findspark.init()
In [2]:
from pyspark.sql import SparkSession

spark = SparkSession \
    .builder \
    .appName("Analyzing airline data") \
    .getOrCreate()
In [3]:
from pyspark.sql.types import Row
from datetime import datetime

Loading in airline data

In [4]:
airlinesPath = "../datasets/airlines.csv"
flightsPath = "../datasets/flights.csv"
airportsPath = "../datasets/airports.csv"
In [5]:
airlines = spark.read\
                .format("csv")\
                .option("header", "true")\
                .load(airlinesPath)
In [6]:
airlines.createOrReplaceTempView("airlines")
In [7]:
airlines = spark.sql("SELECT * FROM airlines")
airlines.columns
Out[7]:
['Code', 'Description']
In [8]:
airlines.show(5)
+-----+--------------------+
| Code|         Description|
+-----+--------------------+
|19031|Mackey Internatio...|
|19032|Munz Northern Air...|
|19033|Cochise Airlines ...|
|19034|Golden Gate Airli...|
|19035|  Aeromech Inc.: RZZ|
+-----+--------------------+
only showing top 5 rows

In [9]:
flights = spark.read\
               .format("csv")\
               .option("header", "true")\
               .load(flightsPath)
In [12]:
flights.createOrReplaceTempView("flights")

flights.columns
Out[12]:
['date',
 'airlines',
 'flight_number',
 'origin',
 'destination',
 'departure',
 'departure_delay',
 'arrival',
 'arrival_delay',
 'air_time',
 'distance']
In [13]:
flights.show(5)
+----------+--------+-------------+------+-----------+---------+---------------+-------+-------------+--------+--------+
|      date|airlines|flight_number|origin|destination|departure|departure_delay|arrival|arrival_delay|air_time|distance|
+----------+--------+-------------+------+-----------+---------+---------------+-------+-------------+--------+--------+
|2014-04-01|   19805|            1|   JFK|        LAX|     0854|          -6.00|   1217|         2.00|  355.00| 2475.00|
|2014-04-01|   19805|            2|   LAX|        JFK|     0944|          14.00|   1736|       -29.00|  269.00| 2475.00|
|2014-04-01|   19805|            3|   JFK|        LAX|     1224|          -6.00|   1614|        39.00|  371.00| 2475.00|
|2014-04-01|   19805|            4|   LAX|        JFK|     1240|          25.00|   2028|       -27.00|  264.00| 2475.00|
|2014-04-01|   19805|            5|   DFW|        HNL|     1300|          -5.00|   1650|        15.00|  510.00| 3784.00|
+----------+--------+-------------+------+-----------+---------+---------------+-------+-------------+--------+--------+
only showing top 5 rows

Counting with dataframes

In [14]:
flights.count(), airlines.count()
Out[14]:
(476881, 1579)

Counting using SQL

In [15]:
flights_count = spark.sql("SELECT COUNT(*) FROM flights")
airlines_count = spark.sql("SELECT COUNT(*) FROM airlines")
In [16]:
flights_count, airlines_count
Out[16]:
(DataFrame[count(1): bigint], DataFrame[count(1): bigint])
In [17]:
flights_count.collect()[0][0], airlines_count.collect()[0][0]
Out[17]:
(476881, 1579)

Dataframes created using SQL commands can be aggregated, grouped etc. exactly as before

In [18]:
total_distance_df = spark.sql("SELECT distance FROM flights")\
                         .agg({"distance":"sum"})\
                         .withColumnRenamed("sum(distance)","total_distance")
In [19]:
total_distance_df.show()
+--------------+
|total_distance|
+--------------+
|  3.79052917E8|
+--------------+

Analyzing flight delays

In [20]:
all_delays_2012 = spark.sql(
    "SELECT date, airlines, flight_number, departure_delay " +
    "FROM flights WHERE departure_delay > 0 and year(date) = 2012")
In [21]:
all_delays_2012.show(5)
+----+--------+-------------+---------------+
|date|airlines|flight_number|departure_delay|
+----+--------+-------------+---------------+
+----+--------+-------------+---------------+

In [23]:
all_delays_2014 = spark.sql(
    "SELECT date, airlines, flight_number, departure_delay " +
    "FROM flights WHERE departure_delay > 0 and year(date) = 2014")

all_delays_2014.show(5)
+----------+--------+-------------+---------------+
|      date|airlines|flight_number|departure_delay|
+----------+--------+-------------+---------------+
|2014-04-01|   19805|            2|          14.00|
|2014-04-01|   19805|            4|          25.00|
|2014-04-01|   19805|            6|         126.00|
|2014-04-01|   19805|            7|         125.00|
|2014-04-01|   19805|            8|           4.00|
+----------+--------+-------------+---------------+
only showing top 5 rows

In [24]:
all_delays_2014.createOrReplaceTempView("all_delays")
In [25]:
all_delays_2014.orderBy(all_delays_2014.departure_delay.desc()).show(5)
+----------+--------+-------------+---------------+
|      date|airlines|flight_number|departure_delay|
+----------+--------+-------------+---------------+
|2014-04-27|   20366|         5246|          99.00|
|2014-04-27|   19393|         2948|          99.00|
|2014-04-27|   20366|         5365|          99.00|
|2014-04-26|   19977|          616|          99.00|
|2014-04-27|   20366|         6030|          99.00|
+----------+--------+-------------+---------------+
only showing top 5 rows

Total number of delayed flights in 2014

In [26]:
delay_count = spark.sql("SELECT COUNT(departure_delay) FROM all_delays")
In [27]:
delay_count.show()
+----------------------+
|count(departure_delay)|
+----------------------+
|                179015|
+----------------------+

In [28]:
delay_count.collect()[0][0]
Out[28]:
179015

Percentage of flights delayed

In [29]:
delay_percent = delay_count.collect()[0][0] / flights_count.collect()[0][0] * 100
delay_percent
Out[29]:
37.53871510922012

Finding delay per aIrlines

In [30]:
delay_per_airline = spark.sql("SELECT airlines, departure_delay FROM flights")\
                         .groupBy("airlines")\
                         .agg({"departure_delay":"avg"})\
                         .withColumnRenamed("avg(departure_delay)", "departure_delay")
In [31]:
delay_per_airline.orderBy(delay_per_airline.departure_delay.desc()).show(5)
+--------+------------------+
|airlines|   departure_delay|
+--------+------------------+
|   19393|13.429567657134724|
|   20366|12.296210112379818|
|   19977| 8.818392620527979|
|   20436| 8.716275167785234|
|   20409|  8.31110357194785|
+--------+------------------+
only showing top 5 rows

In [32]:
delay_per_airline.createOrReplaceTempView("delay_per_airline")
In [33]:
delay_per_airline = spark.sql("SELECT * FROM delay_per_airline ORDER BY departure_delay DESC")
In [34]:
delay_per_airline.show(5)
+--------+------------------+
|airlines|   departure_delay|
+--------+------------------+
|   19393|13.429567657134724|
|   20366|12.296210112379818|
|   19977| 8.818392620527979|
|   20436| 8.716275167785234|
|   20409|  8.31110357194785|
+--------+------------------+
only showing top 5 rows

SQL join operations

  • Get the names of the delayed flights
In [35]:
delay_per_airline = spark.sql("SELECT * FROM delay_per_airline " +
                              "JOIN airlines ON airlines.code = delay_per_airline.airlines " +
                              "ORDER BY departure_delay DESC")

delay_per_airline.show(5)
+--------+------------------+-----+--------------------+
|airlines|   departure_delay| Code|         Description|
+--------+------------------+-----+--------------------+
|   19393|13.429567657134724|19393|Southwest Airline...|
|   20366|12.296210112379818|20366|ExpressJet Airlin...|
|   19977| 8.818392620527979|19977|United Air Lines ...|
|   20436| 8.716275167785234|20436|Frontier Airlines...|
|   20409|  8.31110357194785|20409| JetBlue Airways: B6|
+--------+------------------+-----+--------------------+
only showing top 5 rows

In [36]:
delay_per_airline.drop("code").show(5)
+--------+------------------+--------------------+
|airlines|   departure_delay|         Description|
+--------+------------------+--------------------+
|   19393|13.429567657134724|Southwest Airline...|
|   20366|12.296210112379818|ExpressJet Airlin...|
|   19977| 8.818392620527979|United Air Lines ...|
|   20436| 8.716275167785234|Frontier Airlines...|
|   20409|  8.31110357194785| JetBlue Airways: B6|
+--------+------------------+--------------------+
only showing top 5 rows

In [ ]: