<jupyter_start><jupyter_text>Solutions - Problem 1<jupyter_code>%%HTML
<iframe width="560" height="315" src="https://www.youtube.com/embed/0nyMLUn4TzM?rel=0&amp;controls=1&amp;showinfo=0" frameborder="0" allowfullscreen></iframe><jupyter_output><empty_output><jupyter_text>Get total number of flights as well as number of flights which are delayed in departure and number of flights delayed in arrival.* Output should contain 3 columns - **FlightCount**, **DepDelayedCount**, **ArrDelayedCount** Let us start spark context for this Notebook so that we can execute the code provided. You can sign up for our [10 node state of the art cluster/labs](https://labs.itversity.com/plans) to learn Spark SQL using our unique integrated LMS.<jupyter_code>from pyspark.sql import SparkSession

import getpass
username = getpass.getuser()

spark = SparkSession. \
    builder. \
    config('spark.ui.port', '0'). \
    config("spark.sql.warehouse.dir", f"/user/{username}/warehouse"). \
    enableHiveSupport(). \
    appName(f'{username} | Python - Basic Transformations'). \
    master('yarn'). \
    getOrCreate()<jupyter_output><empty_output><jupyter_text>If you are going to use CLIs, you can use Spark SQL using one of the 3 approaches.**Using Spark SQL**```spark2-sql \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```**Using Scala**```spark2-shell \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```**Using Pyspark**```pyspark2 \    --master yarn \    --conf spark.ui.port=0 \    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse```  Reading airtraffic data<jupyter_code>airtraffic_path = "/public/airtraffic_all/airtraffic-part/flightmonth=200801"
airtraffic = spark. \
    read. \
    parquet(airtraffic_path)
airtraffic.printSchema()<jupyter_output>root
 |-- Year: integer (nullable = true)
 |-- Month: integer (nullable = true)
 |-- DayofMonth: integer (nullable = true)
 |-- DayOfWeek: integer (nullable = true)
 |-- DepTime: string (nullable = true)
 |-- CRSDepTime: integer (nullable = true)
 |-- ArrTime: string (nullable = true)
 |-- CRSArrTime: integer (nullable = true)
 |-- UniqueCarrier: string (nullable = true)
 |-- FlightNum: integer (nullable = true)
 |-- TailNum: string (nullable = true)
 |-- ActualElapsedTime: string (nullable = true)
 |-- CRSElapsedTime: integer (nullable = true)
 |-- AirTime: string (nullable = true)
 |-- ArrDelay: string (nullable = true)
 |-- DepDelay: string (nullable = true)
 |-- Origin: string (nullable = true)
 |-- Dest: string (nullable = true)
 |-- Distance: string (nullable = true)
 |-- TaxiIn: string (nullable = true)
 |-- TaxiOut: string (nullable = true)
 |-- Cancelled: integer (nullable = true)
 |-- CancellationCode: string (nullable = true)
 |-- Diverted: integer (nullable = true)
 |-- Car[...]<jupyter_text>Get flights with delayed arrival<jupyter_code># SQL Style
airtraffic.filter("IsArrDelayed = 'YES' AND Cancelled = 0").show()
# Data Frame Style
airtraffic.filter((airtraffic["IsArrDelayed"] == 'YES') & (airtraffic["Cancelled"] == 0)).show()
airtraffic.filter((airtraffic.IsArrDelayed == 'YES') & (airtraffic.Cancelled == 0)).show()<jupyter_output>+----+-----+----------+---------+-------+----------+-------+----------+-------------+---------+-------+-----------------+--------------+-------+--------+--------+------+----+--------+------+-------+---------+----------------+--------+------------+------------+--------+-------------+-----------------+------------+------------+
|Year|Month|DayofMonth|DayOfWeek|DepTime|CRSDepTime|ArrTime|CRSArrTime|UniqueCarrier|FlightNum|TailNum|ActualElapsedTime|CRSElapsedTime|AirTime|ArrDelay|DepDelay|Origin|Dest|Distance|TaxiIn|TaxiOut|Cancelled|CancellationCode|Diverted|CarrierDelay|WeatherDelay|NASDelay|SecurityDelay|LateAircraftDelay|IsArrDelayed|IsDepDelayed|
+----+-----+----------+---------+-------+----------+-------+----------+-------------+---------+-------+-----------------+--------------+-------+--------+--------+------+----+--------+------+-------+---------+----------------+--------+------------+------------+--------+-------------+-----------------+------------+------------+
|2008|    1|    [...]<jupyter_text>Get delayed counts<jupyter_code>airtraffic. \
    select('IsDepDelayed', 'IsArrDelayed', 'Cancelled'). \
    distinct(). \
    show()
## Departure Delayed Count
airtraffic. \
    filter(airtraffic.IsDepDelayed == "YES"). \
    count()
## Departure Delayed Count
airtraffic. \
    filter((airtraffic.IsDepDelayed == "YES") & (airtraffic.Cancelled == 0)). \
    count()
## Arrival Delayed Count
airtraffic. \
    filter(airtraffic.IsArrDelayed == "YES"). \
    count()
## Arrival Delayed Count
airtraffic. \
    filter((airtraffic.IsArrDelayed == "YES") & (airtraffic.Cancelled == 0)). \
    count()
airtraffic. \
    filter("(IsDepDelayed = 'YES' OR IsArrDelayed = 'YES') AND Cancelled = 0"). \
    select('Year', 'Month', 'DayOfMonth', 
           'FlightNum', 'IsDepDelayed', 'IsArrDelayed'
          ). \
    show()
from pyspark.sql.functions import col, lit, count, sum, expr
## Both Departure Delayed and Arrival Delayed
airtraffic. \
    filter('Cancelled = 0'). \
    agg(count(lit(1)).alias("FlightCount"),
        sum(expr("CASE WHEN IsDepDelayed = 'YES' THEN 1 ELSE 0 END")).alias("DepDelayedCount"),
        sum(expr("CASE WHEN IsArrDelayed = 'YES' THEN 1 ELSE 0 END")).alias("ArrDelayedCount")
       ). \
    show()
from pyspark.sql.functions import when
## Both Departure Delayed and Arrival Delayed
airtraffic. \
    filter('Cancelled = 0'). \
    agg(count(lit(1)).alias("FlightCount"),
        sum(when(col('IsDepDelayed') == 'YES', 1).otherwise(lit(0))).alias("DepDelayedCount"),
        sum(when(col('IsArrDelayed') == lit('YES'), 1).otherwise(lit(0))).alias("ArrDelayedCount")
       ). \
    show()<jupyter_output>+-----------+---------------+---------------+
|FlightCount|DepDelayedCount|ArrDelayedCount|
+-----------+---------------+---------------+
|     588366|         247905|         280663|
+-----------+---------------+---------------+