import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession \
    .builder \
    .appName("Analyzing airline data") \
    .getOrCreate()
from pyspark.sql.types import Row
from datetime import datetime
airlinesPath = "../datasets/airlines.csv"
flightsPath = "../datasets/flights.csv"
airportsPath = "../datasets/airports.csv"
airlines = spark.read\
                .format("csv")\
                .option("header", "true")\
                .load(airlinesPath)
airlines.createOrReplaceTempView("airlines")
airlines = spark.sql("SELECT * FROM airlines")
airlines.columns
airlines.show(5)
flights = spark.read\
               .format("csv")\
               .option("header", "true")\
               .load(flightsPath)
flights.createOrReplaceTempView("flights")
flights.columns
flights.show(5)
flights.count(), airlines.count()
flights_count = spark.sql("SELECT COUNT(*) FROM flights")
airlines_count = spark.sql("SELECT COUNT(*) FROM airlines")
flights_count, airlines_count
flights_count.collect()[0][0], airlines_count.collect()[0][0]
total_distance_df = spark.sql("SELECT distance FROM flights")\
                         .agg({"distance":"sum"})\
                         .withColumnRenamed("sum(distance)","total_distance")
total_distance_df.show()
all_delays_2012 = spark.sql(
    "SELECT date, airlines, flight_number, departure_delay " +
    "FROM flights WHERE departure_delay > 0 and year(date) = 2012")
all_delays_2012.show(5)
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)
all_delays_2014.createOrReplaceTempView("all_delays")
all_delays_2014.orderBy(all_delays_2014.departure_delay.desc()).show(5)
delay_count = spark.sql("SELECT COUNT(departure_delay) FROM all_delays")
delay_count.show()
delay_count.collect()[0][0]
delay_percent = delay_count.collect()[0][0] / flights_count.collect()[0][0] * 100
delay_percent
delay_per_airline = spark.sql("SELECT airlines, departure_delay FROM flights")\
                         .groupBy("airlines")\
                         .agg({"departure_delay":"avg"})\
                         .withColumnRenamed("avg(departure_delay)", "departure_delay")
delay_per_airline.orderBy(delay_per_airline.departure_delay.desc()).show(5)
delay_per_airline.createOrReplaceTempView("delay_per_airline")
delay_per_airline = spark.sql("SELECT * FROM delay_per_airline ORDER BY departure_delay DESC")
delay_per_airline.show(5)
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)
delay_per_airline.drop("code").show(5)