Airline dataset processing with SparkSQL

-

Now let’s explore the Dataset using Spark SQL and DataFrame transformations. After we register the DataFrame as a SQL temporary view, we can use SQL functions on the SparkSession to run SQL queries, which will return the results as a DataFrame. We cache the DataFrame, since we will reuse it and because Spark can cache DataFrames or Tables in columnar format in memory, which can improve memory usage and performance.

// cache DataFrame in columnar format in memorydf.cache// create Table view of DataFrame for Spark SQLdf.createOrReplaceTempView("flights")// cache flights table in columnar format in memoryspark.catalog.cacheTable("flights")

Below, we display information for the top five longest departure delays with Spark SQL and with DataFrame transformations (where a delay is considered greater than 40 minutes):

// Spark SQLspark.sql("select carrier,origin, dest, depdelay,crsdephour, dist, dofW from flights where depdelay > 40 order by depdelay desc limit 5").show// same query using DataFrame transformationsdf.select($"carrier",$"origin",$"dest",$"depdelay", $"crsdephour").filter($"depdelay" > 40).orderBy(desc( "depdelay" )).show(5)---result:---+-------+------+----+--------+----------+|carrier|origin|dest|depdelay|crsdephour|+-------+------+----+--------+----------+|     AA|   SFO| ORD|  1440.0|         8||     DL|   BOS| ATL|  1185.0|        17||     UA|   DEN| EWR|  1138.0|        12||     DL|   ORD| ATL|  1087.0|        19||     UA|   MIA| EWR|  1072.0|        20|+-------+------+----+--------+----------+

Below, we display the average departure delay by carrier:

// DataFrame transformationsdf.groupBy("carrier").agg(avg("depdelay")).show---result:---+-------+------------------+|carrier|     avg(depdelay)|+-------+------------------+|     UA|17.477878450696764||     AA| 10.45768118831622||     DL|15.316061660865241||     WN|13.491000418585182|+-------+-------------------

Let’s explore this data for flight delays, when the departure delay is greater than 40 minutes. Below, we see that United Airlines and Delta have the highest count of flight delays for January and February 2017 (the training set).

// __Count of Departure Delays by Carrier (where delay=40 minutes)__df.filter($"depdelay" > 40).groupBy("carrier").count.orderBy(desc( "count")).show(5)---result:---+-------+-----+|carrier|count|+-------+-----+|     UA| 2420||     DL| 1043||     AA|  757||     WN|  244|+-------+-----+
// Count of Departure Delays by Carrier (where delay=40 minutes)%sqlselect carrier, count(depdelay)from flights where depdelay > 40group by carrier

In the query below, we see that Monday (1), Tuesday (2), and Sunday (7) have the highest count of flight delays.

// Count of Departure Delays by Day of the Week%sqlselect dofW, count(depdelay)from flights where depdelay > 40group by dofW

In the query below, we see that the hours between 13:00-19:00 have the highest count of flight delays.

%sqlselect crsdephour, count(depdelay)from flights where depdelay > 40group by crsdephour order by crsdephour

In the query below, we see that the originating airports, Chicago and Atlanta, have the highest count of flight delays.

%sqlselect origin, count(depdelay)from flights where depdelay > 40group by originORDER BY count(depdelay) desc

In the query below, we see the count of departure delays by origin and destination. The routes ORD->SFO and DEN->SFO have the highest delays, maybe because of weather in January and February. Adding weather to this Dataset would give better results.

%sqlselect origin, dest, count(depdelay)from flights where depdelay > 40group by origin, destORDER BY count(depdelay) desc

Popular posts from this blog

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

Complex SQL: fetch the users who logged in consecutively 3 or more times (lead perfect example)

-
From the login_details table, fetch the users who logged in consecutively 3 or more times. Table Name : LOGIN_DETAILS Approach : We need to fetch users who have appeared 3 or more times consecutively in login details table. There is a window function which can be used to fetch data from the following record. Use that window function to compare the user name in current row with user name in the next row and in the row following the next row. If it matches then fetch those records. –Table Structure: drop table login_details; create table login_details( login_id int primary key, user_name varchar(50) not null, login_date date); delete from login_details; insert into login_details values (101, ‘Michael’, current_date), (102, ‘James’, current_date), (103, ‘Stewart’, current_date+1), (104, ‘Stewart’, current_date+1), (105, ‘Stewart’, current_date+1), (106, ‘Michael’, current_date+2), (107, ‘Michael’, cur...
Read more

Credit Card Data Analysis using PySpark (how to use auto broadcast join after disabling it)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Application: Here we will disable auto broadcast and then join 2 data frames and will see execution plan which should use broadcast join Solution: from pyspark.sql import SparkSession from pyspark.sql.functions import broadcast spark = SparkSession.builder.master(‘local[2]’)\ .appName(‘RDD_Methods_Examples’)\ .getOrCreate() print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”)) spark.conf.set(“spark.sql.autoBroadcastJoinThreshold”, -1) pr...
Read more