DA-Airline_data_pyspark.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "e841b459",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "aa7f0ded",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "import numpy as np\n",
    "import warnings\n",
    "warnings.filterwarnings(\"ignore\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "id": "f7d41eec",
   "metadata": {},
   "outputs": [],
   "source": [
    "#This is another way to extract the files from multiple types of compressed files.\n",
    "#https://docs.python.org/3/library/shutil.html\n",
    "import shutil\n",
    "#shutil.unpack_archive('airLineCapitalOne.zip','newdata')\n",
    "import zipfile\n",
    "\n",
    "with zipfile.ZipFile('airLineCapitalOne.zip','r') as ref:\n",
    "    ref.extractall('airLineData')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "f133e616",
   "metadata": {},
   "outputs": [],
   "source": [
    "metaData = pd.read_excel(\"airLineData/Airline_Challenge_Metadata.xlsx\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "ddbd65a9",
   "metadata": {},
   "outputs": [],
   "source": [
    "flightsPath = \"airLineData/Flights.csv\"\n",
    "ticketsPath = \"airLineData/Tickets.csv\"\n",
    "AirPortCodesPath = \"airLineData/Airport_Codes.csv\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "872c214a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/24 16:09:38 WARN Utils: Your hostname, codeStation resolves to a loopback address: 127.0.1.1; using 192.168.57.83 instead (on interface wlo1)\n",
      "22/11/24 16:09:38 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/24 16:09:41 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    }
   ],
   "source": [
    "#Starting the pyspark session\n",
    "spark = SparkSession.builder.appName(\"CapitalOne DA\").getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "1411408d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------------------+-----------------------------------------------------------------+\n",
      "|key                    |value                                                            |\n",
      "+-----------------------+-----------------------------------------------------------------+\n",
      "|spark.sql.warehouse.dir|file:/run/media/solverbot/repoA/gitFolders/moreDE/spark-warehouse|\n",
      "+-----------------------+-----------------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SET spark.sql.warehouse.dir\").show(truncate=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "4ed74a26",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Creating the local database for storing the data\n",
    "spark.sql(\"CREATE DATABASE IF NOT EXISTS capitalone_DA\")\n",
    "spark.sql(\"USE capitalone_DA\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "60508f35",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Reading the files \n",
    "flights= spark.read.csv(flightsPath,inferSchema=True,header=True)\n",
    "tickets= spark.read.csv(ticketsPath, inferSchema=True, header=True)\n",
    "airport= spark.read.csv(AirPortCodesPath, inferSchema=True, header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "32e5943e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#The tables can be read using read.parquet, if the tables are stored locally\n",
    "airport = spark.read.parquet(\"spark-warehouse/capitalone_da.db/airport/\")\n",
    "tickets = spark.read.parquet(\"spark-warehouse/capitalone_da.db/tickets/\")\n",
    "flights = spark.read.parquet(\"spark-warehouse/capitalone_da.db/flights/\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "12dac942",
   "metadata": {},
   "outputs": [],
   "source": [
    "# creating views in the database\n",
    "flights.createOrReplaceTempView(\"flights\")\n",
    "tickets.createOrReplaceTempView(\"tickets\")\n",
    "airport.createOrReplaceTempView(\"airport\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "36ea0853",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+--------+-----------------+-----------------+------+----------------+---------------+-----------+--------------+---------+---------+---------+--------+--------+--------------+\n",
      "|   FL_DATE|OP_CARRIER|TAIL_NUM|OP_CARRIER_FL_NUM|ORIGIN_AIRPORT_ID|ORIGIN|ORIGIN_CITY_NAME|DEST_AIRPORT_ID|DESTINATION|DEST_CITY_NAME|DEP_DELAY|ARR_DELAY|CANCELLED|AIR_TIME|DISTANCE|OCCUPANCY_RATE|\n",
      "+----------+----------+--------+-----------------+-----------------+------+----------------+---------------+-----------+--------------+---------+---------+---------+--------+--------+--------------+\n",
      "|2019-03-02|        WN|  N955WN|             4591|            14635|   RSW|  Fort Myers, FL|          11042|        CLE| Cleveland, OH|     -8.0|     -6.0|      0.0|   143.0|  1025.0|          0.97|\n",
      "|2019-03-02|        WN|  N8686A|             3231|            14635|   RSW|  Fort Myers, FL|          11066|        CMH|  Columbus, OH|      1.0|      5.0|      0.0|   135.0|   930.0|          0.55|\n",
      "+----------+----------+--------+-----------------+-----------------+------+----------------+---------------+-----------+--------------+---------+---------+---------+--------+--------+--------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "9e2c1894",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----------+\n",
      "|namespace|tableName|isTemporary|\n",
      "+---------+---------+-----------+\n",
      "|         |  airport|       true|\n",
      "|         |  flights|       true|\n",
      "|         |  tickets|       true|\n",
      "+---------+---------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Temporary tables of the above data is created in the spark execution environment\n",
    "spark.sql(\"SHOW TABLES\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "0fa5fae6",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- FL_DATE: string (nullable = true)\n",
      " |-- OP_CARRIER: string (nullable = true)\n",
      " |-- TAIL_NUM: string (nullable = true)\n",
      " |-- OP_CARRIER_FL_NUM: string (nullable = true)\n",
      " |-- ORIGIN_AIRPORT_ID: integer (nullable = true)\n",
      " |-- ORIGIN: string (nullable = true)\n",
      " |-- ORIGIN_CITY_NAME: string (nullable = true)\n",
      " |-- DEST_AIRPORT_ID: integer (nullable = true)\n",
      " |-- DESTINATION: string (nullable = true)\n",
      " |-- DEST_CITY_NAME: string (nullable = true)\n",
      " |-- DEP_DELAY: double (nullable = true)\n",
      " |-- ARR_DELAY: double (nullable = true)\n",
      " |-- CANCELLED: double (nullable = true)\n",
      " |-- AIR_TIME: string (nullable = true)\n",
      " |-- DISTANCE: string (nullable = true)\n",
      " |-- OCCUPANCY_RATE: double (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "f225071c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- TYPE: string (nullable = true)\n",
      " |-- NAME: string (nullable = true)\n",
      " |-- ELEVATION_FT: double (nullable = true)\n",
      " |-- CONTINENT: string (nullable = true)\n",
      " |-- ISO_COUNTRY: string (nullable = true)\n",
      " |-- MUNICIPALITY: string (nullable = true)\n",
      " |-- IATA_CODE: string (nullable = true)\n",
      " |-- COORDINATES: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airport.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "a2b0b7f9",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- ITIN_ID: long (nullable = true)\n",
      " |-- YEAR: integer (nullable = true)\n",
      " |-- QUARTER: integer (nullable = true)\n",
      " |-- ORIGIN: string (nullable = true)\n",
      " |-- ORIGIN_COUNTRY: string (nullable = true)\n",
      " |-- ORIGIN_STATE_ABR: string (nullable = true)\n",
      " |-- ORIGIN_STATE_NM: string (nullable = true)\n",
      " |-- ROUNDTRIP: double (nullable = true)\n",
      " |-- REPORTING_CARRIER: string (nullable = true)\n",
      " |-- PASSENGERS: double (nullable = true)\n",
      " |-- ITIN_FARE: string (nullable = true)\n",
      " |-- DESTINATION: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "tickets.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "3f501a45",
   "metadata": {},
   "outputs": [],
   "source": [
    "#The df.sql function APIs are helpful \n",
    "flights_condensed = flights.select(date_format(col('FL_DATE'),'y-MM-dd').alias('flight_date'),\n",
    "                'OP_CARRIER','TAIL_NUM','OP_CARRIER_FL_NUM','ORIGIN','DESTINATION',\n",
    "               'DEP_DELAY','ARR_DELAY','CANCELLED','AIR_TIME','DISTANCE',\n",
    "               'OCCUPANCY_RATE')\n",
    "flights_condensed.createOrReplaceTempView(\"flightCondensed\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "id": "22d746ec",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+--------+-----------------+------+-----------+---------+---------+---------+--------+--------+--------------+\n",
      "|flight_date|OP_CARRIER|TAIL_NUM|OP_CARRIER_FL_NUM|ORIGIN|DESTINATION|DEP_DELAY|ARR_DELAY|CANCELLED|AIR_TIME|DISTANCE|OCCUPANCY_RATE|\n",
      "+-----------+----------+--------+-----------------+------+-----------+---------+---------+---------+--------+--------+--------------+\n",
      "| 2019-03-02|        WN|  N955WN|             4591|   RSW|        CLE|     -8.0|     -6.0|      0.0|   143.0|  1025.0|          0.97|\n",
      "| 2019-03-02|        WN|  N8686A|             3231|   RSW|        CMH|      1.0|      5.0|      0.0|   135.0|   930.0|          0.55|\n",
      "| 2019-03-02|        WN|  N201LV|             3383|   RSW|        CMH|      0.0|      4.0|      0.0|   132.0|   930.0|          0.91|\n",
      "| 2019-03-02|        WN|  N413WN|             5498|   RSW|        CMH|     11.0|     14.0|      0.0|   136.0|   930.0|          0.67|\n",
      "| 2019-03-02|        WN|  N7832A|             6933|   RSW|        DAL|      0.0|    -17.0|      0.0|   151.0|  1005.0|          0.62|\n",
      "+-----------+----------+--------+-----------------+------+-----------+---------+---------+---------+--------+--------+--------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT * FROM flightCondensed LIMIT 5\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "698ac362",
   "metadata": {},
   "outputs": [],
   "source": [
    "airportCondensed = spark.sql(\"\"\"SELECT TYPE, NAME, ELEVATION_FT, MUNICIPALITY, IATA_CODE, COORDINATES FROM airport\"\"\")\n",
    "airportCondensed.createOrReplaceTempView(\"airportCondensed\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "76d252f3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------------+------------+------------+---------+--------------------+\n",
      "|         TYPE|                NAME|ELEVATION_FT|MUNICIPALITY|IATA_CODE|         COORDINATES|\n",
      "+-------------+--------------------+------------+------------+---------+--------------------+\n",
      "|     heliport|   Total Rf Heliport|        11.0|    Bensalem|     null|-74.9336013793945...|\n",
      "|small_airport|Aero B Ranch Airport|      3435.0|       Leoti|     null|-101.473911, 38.7...|\n",
      "|small_airport|        Lowell Field|       450.0|Anchor Point|     null|-151.695999146, 5...|\n",
      "|small_airport|        Epps Airpark|       820.0|     Harvest|     null|-86.7703018188476...|\n",
      "|       closed|Newport Hospital ...|       237.0|     Newport|     null| -91.254898, 35.6087|\n",
      "+-------------+--------------------+------------+------------+---------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT * FROM airportCondensed\"\"\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "c940da42",
   "metadata": {},
   "outputs": [],
   "source": [
    "ticketsCondensed = spark.sql(\"\"\"SELECT ITIN_ID, YEAR, QUARTER, ORIGIN, ORIGIN_STATE_ABR, DESTINATION,\n",
    "              ROUNDTRIP, REPORTING_CARRIER, PASSENGERS, ITIN_FARE FROM tickets\"\"\")\n",
    "ticketsCondensed.createOrReplaceTempView(\"ticketsCondensed\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "id": "95e0747f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+----------------+-----------+\n",
      "|namespace|       tableName|isTemporary|\n",
      "+---------+----------------+-----------+\n",
      "|         |         airport|       true|\n",
      "|         |airportcondensed|       true|\n",
      "|         | flightcondensed|       true|\n",
      "|         |         flights|       true|\n",
      "|         |         tickets|       true|\n",
      "|         |ticketscondensed|       true|\n",
      "+---------+----------------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SHOW TABLES\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f16a3c53",
   "metadata": {},
   "source": [
    "Analytics questions on the Airline data\n",
    "0) On each date, how many flights are there\n",
    "1) How many unique flights are there in the datasets\n",
    "2) Which flights have travelled the longest distance\n",
    "3) Which routes has the maximum flights\n",
    "4) Which routes has the maximum number of round trips\n",
    "5) What is the highest elevation of the airport\n",
    "6) Which route has the maximum revenue\n",
    "7) Which route has the maximum number of passenger\n",
    "8) What is the number of passenger carried by the aircraft till date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "b2c89526",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 26:==============>                                           (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------------+\n",
      "|flight_date|flight_counts|\n",
      "+-----------+-------------+\n",
      "| 2019-03-15|        23361|\n",
      "| 2019-03-14|        23305|\n",
      "| 2019-03-22|        23258|\n",
      "| 2019-03-29|        23250|\n",
      "| 2019-03-11|        23249|\n",
      "+-----------+-------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#What is the number of flights operating on each date\n",
    "\n",
    "spark.sql(\"\"\"SELECT flight_date, COUNT(flight_date) AS flight_counts \n",
    "            FROM flightCondensed\n",
    "            GROUP BY flight_date\n",
    "            ORDER BY COUNT(flight_date) DESC\"\"\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "id": "bbf54e37",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+----------+\n",
      "|TAIL_NUM|tripCounts|\n",
      "+--------+----------+\n",
      "|  N485HA|       928|\n",
      "|  N479HA|       882|\n",
      "|  N483HA|       882|\n",
      "|  N480HA|       875|\n",
      "|  N491HA|       873|\n",
      "+--------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# How many unique flights are there in the datasets\n",
    "spark.sql(\"\"\"SELECT TAIL_NUM, \n",
    "                COUNT(TAIL_NUM) AS tripCounts\n",
    "                FROM flightCondensed\n",
    "                GROUP BY TAIL_NUM\n",
    "                ORDER BY COUNT(TAIL_NUM) DESC\"\"\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "d80a74b3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+\n",
      "|OP_CARRIER|tripCounts|\n",
      "+----------+----------+\n",
      "|        WN|    326093|\n",
      "|        AA|    232403|\n",
      "|        DL|    225391|\n",
      "|        OO|    194934|\n",
      "|        UA|    142826|\n",
      "+----------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# How many trips each operator has done\n",
    "spark.sql(\"\"\"SELECT OP_CARRIER,\n",
    "                COUNT(TAIL_NUM) AS tripCounts\n",
    "                FROM flightCondensed\n",
    "                GROUP BY OP_CARRIER\n",
    "                ORDER BY COUNT(TAIL_NUM) DESC\"\"\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "id": "b5f120b7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "26"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#How many different carriers are there in the datasets?\n",
    "spark.sql(\"\"\"SELECT OP_CARRIER,\n",
    "                COUNT(TAIL_NUM) AS tripCounts\n",
    "                FROM flightCondensed\n",
    "                GROUP BY OP_CARRIER\n",
    "                ORDER BY COUNT(TAIL_NUM) DESC\"\"\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "8386c758",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|TAIL_NUM|\n",
      "+--------+\n",
      "|  N8554X|\n",
      "|  N8322X|\n",
      "|  N954WN|\n",
      "|  N8513F|\n",
      "|  N445WN|\n",
      "|  N8525S|\n",
      "|  N914WN|\n",
      "|  N8511K|\n",
      "|  N251WN|\n",
      "|  N496WN|\n",
      "|  N8734Q|\n",
      "|  N448WN|\n",
      "|  N240WN|\n",
      "|  N430WN|\n",
      "|  N8634A|\n",
      "|  N757LV|\n",
      "|  N7838A|\n",
      "|  N7715E|\n",
      "|  N8679A|\n",
      "|  N404WN|\n",
      "+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# How many flight numbers are there under each flight numbers\n",
    "spark.sql(\"\"\"SELECT DISTINCT TAIL_NUM\n",
    "              FROM flightCondensed\n",
    "              WHERE OP_CARRIER = 'WN'\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "e24c987f",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|TAIL_NUM|\n",
      "+--------+\n",
      "|  N8554X|\n",
      "|  N8322X|\n",
      "|  N954WN|\n",
      "|  N8513F|\n",
      "|  N445WN|\n",
      "|  N8525S|\n",
      "|  N914WN|\n",
      "|  N8511K|\n",
      "|  N251WN|\n",
      "|  N496WN|\n",
      "|  N8734Q|\n",
      "|  N448WN|\n",
      "|  N240WN|\n",
      "|  N430WN|\n",
      "|  N8634A|\n",
      "|  N757LV|\n",
      "|  N7838A|\n",
      "|  N7715E|\n",
      "|  N8679A|\n",
      "|  N404WN|\n",
      "+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# How many flight numbers are there under each flight numbers\n",
    "spark.sql(\"\"\"SELECT DISTINCT TAIL_NUM\n",
    "              FROM flightCondensed\n",
    "              WHERE OP_CARRIER = 'WN'\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "id": "bbfca566",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 62:=============================>                            (2 + 2) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------+\n",
      "|TOTAL_DELAY|TAIL_NUM|\n",
      "+-----------+--------+\n",
      "|    19178.0|  N927SW|\n",
      "|    16269.0|  N786SK|\n",
      "|    15404.0|  N14143|\n",
      "|    15099.0|  N920SW|\n",
      "|    15015.0|  N21197|\n",
      "|    14694.0|  N908EV|\n",
      "|    14103.0|  N11191|\n",
      "|    14051.0|  N954SW|\n",
      "|    13938.0|  N780SK|\n",
      "|    13888.0|  N758EV|\n",
      "|    13885.0|  N690CA|\n",
      "|    13843.0|  N693BR|\n",
      "|    13755.0|  N880AS|\n",
      "|    13672.0|  N466SW|\n",
      "|    13518.0|  N788SK|\n",
      "|    13352.0|  N134SY|\n",
      "|    13292.0|  N760SK|\n",
      "|    13198.0|  N871AS|\n",
      "|    13173.0|  N331CA|\n",
      "|    13172.0|  N791SK|\n",
      "+-----------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#Which flight numbers are getting delayed frequently\n",
    "spark.sql(\"\"\"SELECT SUM(DEP_DELAY) AS TOTAL_DELAY, TAIL_NUM\n",
    "            FROM flightCondensed\n",
    "            GROUP BY TAIL_NUM\n",
    "            ORDER BY SUM(DEP_DELAY) DESC\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "id": "3ca7f58c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+\n",
      "|TOTAL_DELAY|OP_CARRIER|\n",
      "+-----------+----------+\n",
      "|  3179490.0|        WN|\n",
      "|  3014699.0|        OO|\n",
      "|  2251096.0|        AA|\n",
      "|  1785352.0|        DL|\n",
      "|  1708696.0|        UA|\n",
      "|  1308290.0|        B6|\n",
      "|   686371.0|        MQ|\n",
      "|   623717.0|        9E|\n",
      "|   603639.0|        YX|\n",
      "|   590527.0|        OH|\n",
      "|   569127.0|        YV|\n",
      "|   520857.0|        EV|\n",
      "|   401598.0|        G7|\n",
      "|   389617.0|        NK|\n",
      "|   361070.0|        F9|\n",
      "|   350897.0|        AX|\n",
      "|   348663.0|        AS|\n",
      "|   335448.0|        C5|\n",
      "|   326916.0|        CP|\n",
      "|   243367.0|        ZW|\n",
      "+-----------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#What is the total departure delays of the Op_carriers\n",
    "spark.sql(\"\"\"SELECT SUM(DEP_DELAY) AS TOTAL_DELAY, OP_CARRIER\n",
    "            FROM flightCondensed\n",
    "            GROUP BY OP_CARRIER\n",
    "            ORDER BY SUM(DEP_DELAY) DESC\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "id": "c57fde30",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+--------+\n",
      "|TOTAL_DELAY|TAIL_NUM|\n",
      "+-----------+--------+\n",
      "|    17831.0|  N927SW|\n",
      "|    15995.0|  N786SK|\n",
      "|    14752.0|  N14143|\n",
      "|    14664.0|  N21197|\n",
      "|    14096.0|  N920SW|\n",
      "|    13809.0|  N908EV|\n",
      "|    13369.0|  N780SK|\n",
      "|    13311.0|  N788SK|\n",
      "|    13231.0|  N11191|\n",
      "|    13060.0|  N693BR|\n",
      "|    12944.0|  N954SW|\n",
      "|    12920.0|  N880AS|\n",
      "|    12713.0|  N758EV|\n",
      "|    12712.0|  N760SK|\n",
      "|    12607.0|  N134SY|\n",
      "|    12543.0|  N466SW|\n",
      "|    12371.0|  N18120|\n",
      "|    12348.0|  N791SK|\n",
      "|    12254.0|  N782SK|\n",
      "|    12130.0|  N14204|\n",
      "+-----------+--------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Which flight numbers are getting delayed frequently\n",
    "spark.sql(\"\"\"SELECT SUM(ARR_DELAY) AS TOTAL_DELAY, TAIL_NUM\n",
    "            FROM flightCondensed\n",
    "            GROUP BY TAIL_NUM\n",
    "            ORDER BY SUM(ARR_DELAY) DESC\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "id": "1ebea516",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+\n",
      "|TOTAL_DELAY|OP_CARRIER|\n",
      "+-----------+----------+\n",
      "|  2173234.0|        OO|\n",
      "|  1228558.0|        AA|\n",
      "|  1017167.0|        UA|\n",
      "|   893889.0|        B6|\n",
      "|   750513.0|        WN|\n",
      "|   559109.0|        MQ|\n",
      "|   457229.0|        EV|\n",
      "|   454123.0|        YV|\n",
      "|   354137.0|        YX|\n",
      "|   313872.0|        C5|\n",
      "|   311863.0|        AX|\n",
      "|   275975.0|        G7|\n",
      "|   254221.0|        OH|\n",
      "|   242416.0|        CP|\n",
      "|   205151.0|        AS|\n",
      "|   191928.0|        9E|\n",
      "|   177492.0|        F9|\n",
      "|   175002.0|        QX|\n",
      "|   157844.0|        ZW|\n",
      "|   147277.0|        G4|\n",
      "+-----------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#What is the total departure delays of the Op_carriers\n",
    "spark.sql(\"\"\"SELECT SUM(ARR_DELAY) AS TOTAL_DELAY, OP_CARRIER\n",
    "            FROM flightCondensed\n",
    "            GROUP BY OP_CARRIER\n",
    "            ORDER BY SUM(ARR_DELAY) DESC\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "f646b399",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(flight_date='2019-03-22', DEP_TOTAL_DELAY=258541.0, ARR_TOTAL_DELAY=133729.0),\n",
       " Row(flight_date='2019-03-23', DEP_TOTAL_DELAY=116769.0, ARR_TOTAL_DELAY=-34271.0),\n",
       " Row(flight_date='2019-03-24', DEP_TOTAL_DELAY=144381.0, ARR_TOTAL_DELAY=-27920.0),\n",
       " Row(flight_date='2019-03-25', DEP_TOTAL_DELAY=183577.0, ARR_TOTAL_DELAY=47952.0),\n",
       " Row(flight_date='2019-03-26', DEP_TOTAL_DELAY=75968.0, ARR_TOTAL_DELAY=-76875.0),\n",
       " Row(flight_date='2019-03-27', DEP_TOTAL_DELAY=117087.0, ARR_TOTAL_DELAY=-39843.0),\n",
       " Row(flight_date='2019-03-28', DEP_TOTAL_DELAY=95720.0, ARR_TOTAL_DELAY=-69509.0),\n",
       " Row(flight_date='2019-03-29', DEP_TOTAL_DELAY=103217.0, ARR_TOTAL_DELAY=-57336.0),\n",
       " Row(flight_date='2019-03-30', DEP_TOTAL_DELAY=69542.0, ARR_TOTAL_DELAY=-74062.0),\n",
       " Row(flight_date='2019-03-31', DEP_TOTAL_DELAY=135289.0, ARR_TOTAL_DELAY=-3327.0)]"
      ]
     },
     "execution_count": 60,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#How is the departure and arrival delays on each date?\n",
    "spark.sql(\"\"\"SELECT flight_date, SUM(DEP_DELAY) AS DEP_TOTAL_DELAY,\n",
    "                    SUM(ARR_DELAY) AS ARR_TOTAL_DELAY\n",
    "                    FROM flightCondensed\n",
    "                    GROUP BY flight_date\n",
    "                    ORDER BY flight_date\"\"\").tail(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "a82ca4af",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "[Row(flight_date=None, DEP_TOTAL_DELAY=30560.0, ARR_TOTAL_DELAY=3950.0),\n",
       " Row(flight_date='2019-01-01', DEP_TOTAL_DELAY=232562.0, ARR_TOTAL_DELAY=128217.0),\n",
       " Row(flight_date='2019-01-02', DEP_TOTAL_DELAY=291099.0, ARR_TOTAL_DELAY=193013.0),\n",
       " Row(flight_date='2019-01-03', DEP_TOTAL_DELAY=193078.0, ARR_TOTAL_DELAY=56174.0),\n",
       " Row(flight_date='2019-01-04', DEP_TOTAL_DELAY=110566.0, ARR_TOTAL_DELAY=-52259.0),\n",
       " Row(flight_date='2019-01-05', DEP_TOTAL_DELAY=159352.0, ARR_TOTAL_DELAY=30358.0),\n",
       " Row(flight_date='2019-01-06', DEP_TOTAL_DELAY=259492.0, ARR_TOTAL_DELAY=120118.0),\n",
       " Row(flight_date='2019-01-07', DEP_TOTAL_DELAY=135680.0, ARR_TOTAL_DELAY=-5475.0),\n",
       " Row(flight_date='2019-01-08', DEP_TOTAL_DELAY=65511.0, ARR_TOTAL_DELAY=-82616.0),\n",
       " Row(flight_date='2019-01-09', DEP_TOTAL_DELAY=45445.0, ARR_TOTAL_DELAY=-92416.0)]"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT flight_date, SUM(DEP_DELAY) AS DEP_TOTAL_DELAY,\n",
    "                    SUM(ARR_DELAY) AS ARR_TOTAL_DELAY\n",
    "                    FROM flightCondensed\n",
    "                    GROUP BY flight_date\n",
    "                    ORDER BY flight_date\"\"\").head(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "id": "34c4ef99",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 9:=============================>                             (2 + 2) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+---------+--------+\n",
      "|flight_date|OP_CARRIER|DEP_DELAY|TAIL_NUM|\n",
      "+-----------+----------+---------+--------+\n",
      "| 2019-01-08|        AA|   1281.0|  N905NN|\n",
      "| 2019-01-08|        OO|   1014.0|  N805SK|\n",
      "+-----------+----------+---------+--------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "#On each date, what is the delay of each op_carrier\n",
    "spark.sql(\"\"\"SELECT flight_date, OP_CARRIER,DEP_DELAY,TAIL_NUM\n",
    "                FROM flightCondensed\n",
    "                WHERE flight_date = '2019-01-08'\n",
    "                ORDER BY DEP_DELAY DESC\"\"\").show(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "51a44a9c",
   "metadata": {},
   "source": [
    "- Get number of flights departed from each of the US airport.\n",
    "- Get number of flights departed from each of the state.\n",
    "- Get the list of airports in the US from which flights are not departed.\n",
    "- Check if there are any origins in airtraffic data which do not have record in airport-codes.\n",
    "- Get the total number of flights from the airports that do not contain entries in airport-codes.\n",
    "- Get the total number of flights per airport that do not contain entries in airport-codes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "ac2cb2ca",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------+----+-------+------+--------------+----------------+---------------+---------+-----------------+----------+---------+-----------+\n",
      "|     ITIN_ID|YEAR|QUARTER|ORIGIN|ORIGIN_COUNTRY|ORIGIN_STATE_ABR|ORIGIN_STATE_NM|ROUNDTRIP|REPORTING_CARRIER|PASSENGERS|ITIN_FARE|DESTINATION|\n",
      "+------------+----+-------+------+--------------+----------------+---------------+---------+-----------------+----------+---------+-----------+\n",
      "|201912723049|2019|      1|   ABI|            US|              TX|          Texas|      1.0|               MQ|       1.0|    736.0|        DAB|\n",
      "|201912723085|2019|      1|   ABI|            US|              TX|          Texas|      1.0|               MQ|       1.0|    570.0|        COS|\n",
      "|201912723491|2019|      1|   ABI|            US|              TX|          Texas|      1.0|               MQ|       1.0|    564.0|        MCO|\n",
      "|201912723428|2019|      1|   ABI|            US|              TX|          Texas|      1.0|               MQ|       1.0|    345.0|        LGA|\n",
      "|201912723509|2019|      1|   ABI|            US|              TX|          Texas|      0.0|               MQ|       1.0|    309.0|        MGM|\n",
      "+------------+----+-------+------+--------------+----------------+---------------+---------+-----------------+----------+---------+-----------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "#Number of flights departed from US Airport\n",
    "spark.sql(\"\"\"SELECT * FROM tickets \n",
    "                WHERE ORIGIN_COUNTRY = 'US' \n",
    "                LIMIT 5\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "22468ea1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 41:=============================>                            (2 + 2) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+\n",
      "|count(1)|\n",
      "+--------+\n",
      "| 1167285|\n",
      "+--------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT COUNT(1) FROM tickets \n",
    "                WHERE ORIGIN_COUNTRY = 'US'\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "id": "9c0f249b",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "1094911"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT DISTINCT ITIN_ID FROM tickets \n",
    "                WHERE ORIGIN_COUNTRY = 'US'\"\"\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "c450d841",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 23:=============================>                            (2 + 2) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|ORIGIN_COUNTRY|\n",
      "+--------------+\n",
      "|            US|\n",
      "+--------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT DISTINCT ORIGIN_COUNTRY FROM tickets\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "id": "d22982d2",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"\"\"CREATE OR REPLACE TEMP VIEW stateFlights AS\n",
    "            SELECT COUNT(1) AS flights, ORIGIN_STATE_NM\n",
    "                FROM tickets\n",
    "                GROUP BY ORIGIN_STATE_NM\n",
    "            ORDER BY COUNT(1) DESC\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "id": "07c40e8b",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 42:==============>                                           (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+---------------+\n",
      "|flights|ORIGIN_STATE_NM|\n",
      "+-------+---------------+\n",
      "| 129910|     California|\n",
      "|  97566|          Texas|\n",
      "|  97179|        Florida|\n",
      "|  65175|       New York|\n",
      "|  44276|       Illinois|\n",
      "|  41030|       Virginia|\n",
      "|  37096|     Washington|\n",
      "|  37041| North Carolina|\n",
      "|  35039|   Pennsylvania|\n",
      "|  34599|       Michigan|\n",
      "|  32208|       Colorado|\n",
      "|  31796|        Georgia|\n",
      "|  27981|      Minnesota|\n",
      "|  27657|        Arizona|\n",
      "|  26377|           Ohio|\n",
      "|  25559|       Missouri|\n",
      "|  25499|  Massachusetts|\n",
      "|  23847|     New Jersey|\n",
      "|  22494|      Tennessee|\n",
      "|  20800|         Oregon|\n",
      "+-------+---------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM stateFlights\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "1216c67a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 30:==============>                                           (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+-----------+\n",
      "|flights|ORIGIN|DESTINATION|\n",
      "+-------+------+-----------+\n",
      "|   1475|   JFK|        LAX|\n",
      "|   1326|   LAX|        JFK|\n",
      "|   1245|   JFK|        SFO|\n",
      "|   1183|   ORD|        LGA|\n",
      "|   1171|   LGA|        ORD|\n",
      "|   1109|   ORD|        LAX|\n",
      "|   1053|   SFO|        LAX|\n",
      "|   1045|   EWR|        SFO|\n",
      "|   1025|   LAX|        SFO|\n",
      "|   1021|   LGA|        MIA|\n",
      "|   1004|   MSP|        PHX|\n",
      "|    963|   EWR|        FLL|\n",
      "|    961|   BOS|        SFO|\n",
      "|    953|   LAX|        HNL|\n",
      "|    953|   BOS|        MCO|\n",
      "|    948|   ORD|        SFO|\n",
      "|    943|   LAX|        LAS|\n",
      "|    941|   LAX|        MCO|\n",
      "|    936|   SFO|        JFK|\n",
      "|    930|   BOS|        LAX|\n",
      "+-------+------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT COUNT(1) AS flights, ORIGIN, DESTINATION \n",
    "                FROM tickets\n",
    "                GROUP BY ORIGIN, DESTINATION\n",
    "            ORDER BY COUNT(1) DESC\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "id": "93c66c70",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 28,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#The view has to be \"temp\" view since the source table views are temporary\n",
    "spark.sql(\"\"\"CREATE OR REPLACE TEMP VIEW flightConnections AS\n",
    "            SELECT COUNT(1) AS flights, ORIGIN, DESTINATION \n",
    "                FROM tickets\n",
    "                GROUP BY ORIGIN, DESTINATION\n",
    "            ORDER BY COUNT(1) DESC\"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "c9947346",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 33:==============>                                           (1 + 3) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+\n",
      "|totalFlight|\n",
      "+-----------+\n",
      "|    1167285|\n",
      "+-----------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT SUM(flights) AS totalFlight FROM flightConnections\"\"\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "21be80d1",
   "metadata": {},
   "source": [
    "- Get the list of airports in the US from which flights are not departed.\n",
    "- Check if there are any origins in airtraffic data which do not have record in airport-codes.\n",
    "- Get the total number of flights from the airports that do not contain entries in airport-codes.\n",
    "- Get the total number of flights per airport that do not contain entries in airport-codes."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "id": "d4735205",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame[]"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark.sql(\"\"\"CREATE OR REPLACE TEMP VIEW airportIATACodes AS\n",
    "            SELECT * FROM airportcondensed\n",
    "              WHERE IATA_CODE IS NOT NULL \"\"\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "id": "8a58c437",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+--------------------+------------+-------------+---------+--------------------+\n",
      "|         TYPE|                NAME|ELEVATION_FT| MUNICIPALITY|IATA_CODE|         COORDINATES|\n",
      "+-------------+--------------------+------------+-------------+---------+--------------------+\n",
      "|small_airport|      Utirik Airport|         4.0|Utirik Island|      UTK|  169.852005, 11.222|\n",
      "|small_airport|Ocean Reef Club A...|         8.0|    Key Largo|      OCA|-80.274803161621,...|\n",
      "|small_airport|Pilot Station Air...|       305.0|Pilot Station|      PQS|-162.899994, 61.9...|\n",
      "|small_airport|Crested Butte Air...|      8980.0|Crested Butte|      CSE|-106.928341, 38.8...|\n",
      "|small_airport|   LBJ Ranch Airport|      1515.0| Johnson City|      JCY|-98.6224975585999...|\n",
      "+-------------+--------------------+------------+-------------+---------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT * FROM airportIATACodes\"\"\").show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "id": "f98a7a78",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 72:=============>    (3 + 1) / 4][Stage 73:=========>        (1 + 1) / 2]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+--------+-------------+--------------------+---------+\n",
      "|FL_DATE|OP_CARRIER|TAIL_NUM|         TYPE|                NAME|IATA_CODE|\n",
      "+-------+----------+--------+-------------+--------------------+---------+\n",
      "|   null|      null|    null|small_airport|SÃ£o Pedro da Ald...|        0|\n",
      "|   null|      null|    null|small_airport|EMBRAER - Unidade...|        0|\n",
      "|   null|      null|    null|small_airport| Lagoa Santa Airport|        0|\n",
      "|   null|      null|    null|small_airport|Fazenda Santa Mar...|        0|\n",
      "|   null|      null|    null|small_airport|Usina Catanduva A...|        0|\n",
      "|   null|      null|    null|small_airport|Fazenda ConstÃ¢nc...|        0|\n",
      "|   null|      null|    null|small_airport|      GarÃ§a Airport|        0|\n",
      "|   null|      null|    null|small_airport| Jaboticabal Airport|        0|\n",
      "|   null|      null|    null|small_airport|SÃ£o Joaquim da B...|        0|\n",
      "|   null|      null|    null|small_airport|Fazenda Barra do ...|        0|\n",
      "|   null|      null|    null|small_airport|    LucÃ©lia Airport|        0|\n",
      "|   null|      null|    null|small_airport|        Leme Airport|        0|\n",
      "|   null|      null|    null|small_airport|      MatÃ£o Airport|        0|\n",
      "|   null|      null|    null|small_airport|    Mirassol Airport|        0|\n",
      "|   null|      null|    null|small_airport|  Mogi Mirim Airport|        0|\n",
      "|   null|      null|    null|small_airport|Novo Horizonte Ai...|        0|\n",
      "|   null|      null|    null|small_airport|Fazenda SÃ£o Fran...|        0|\n",
      "|   null|      null|    null|small_airport|Presidente Vences...|        0|\n",
      "|   null|      null|    null|small_airport|Fazenda das Repre...|        0|\n",
      "|   null|      null|    null|small_airport|Fazenda SÃ£o JoÃ£...|        0|\n",
      "+-------+----------+--------+-------------+--------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 76:>                                                         (0 + 1) / 1]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT f.FL_DATE, f.OP_CARRIER, f.TAIL_NUM,\n",
    "                a.TYPE, a.NAME, a.IATA_CODE\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE \n",
    "                WHERE FL_DATE IS NULL\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "id": "41b272cc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "362"
      ]
     },
     "execution_count": 53,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Number of Airports flight departing\n",
    "spark.sql(\"\"\"SELECT DISTINCT a.NAME\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE \n",
    "                WHERE f.TAIL_NUM IS NOT NULL\"\"\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "id": "928d6135",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "data": {
      "text/plain": [
       "8972"
      ]
     },
     "execution_count": 54,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Number of Airports flights not departing\n",
    "spark.sql(\"\"\"SELECT DISTINCT a.NAME\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE \n",
    "                WHERE f.TAIL_NUM IS NULL\"\"\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "id": "441c6bcb",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "9102"
      ]
     },
     "execution_count": 55,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "#Total Number of Airports\n",
    "spark.sql(\"\"\"SELECT DISTINCT a.NAME\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE\"\"\").count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "id": "4501aa58",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 144:>                                                        (0 + 4) / 4]\r",
      "\r",
      "[Stage 144:>                (0 + 4) / 4][Stage 145:>                (0 + 0) / 2]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+--------+----+----+---------+\n",
      "|FL_DATE|OP_CARRIER|TAIL_NUM|TYPE|NAME|IATA_CODE|\n",
      "+-------+----------+--------+----+----+---------+\n",
      "+-------+----------+--------+----+----+---------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 144:==========================================>              (3 + 1) / 4]\r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT f.FL_DATE, f.OP_CARRIER, f.TAIL_NUM,\n",
    "                a.TYPE, a.NAME, a.IATA_CODE\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE \n",
    "                WHERE a.IATA_CODE IS NULL\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "54d98d98",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 156:============================>                            (2 + 2) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+--------+------+-------------+--------------------+---------+\n",
      "|FL_DATE|OP_CARRIER|TAIL_NUM|ORIGIN|         TYPE|                NAME|IATA_CODE|\n",
      "+-------+----------+--------+------+-------------+--------------------+---------+\n",
      "|   null|      null|    null|  null|small_airport|SÃ£o Pedro da Ald...|        0|\n",
      "|   null|      null|    null|  null|small_airport|EMBRAER - Unidade...|        0|\n",
      "|   null|      null|    null|  null|small_airport| Lagoa Santa Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport|Fazenda Santa Mar...|        0|\n",
      "|   null|      null|    null|  null|small_airport|Usina Catanduva A...|        0|\n",
      "|   null|      null|    null|  null|small_airport|Fazenda ConstÃ¢nc...|        0|\n",
      "|   null|      null|    null|  null|small_airport|      GarÃ§a Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport| Jaboticabal Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport|SÃ£o Joaquim da B...|        0|\n",
      "|   null|      null|    null|  null|small_airport|Fazenda Barra do ...|        0|\n",
      "|   null|      null|    null|  null|small_airport|    LucÃ©lia Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport|        Leme Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport|      MatÃ£o Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport|    Mirassol Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport|  Mogi Mirim Airport|        0|\n",
      "|   null|      null|    null|  null|small_airport|Novo Horizonte Ai...|        0|\n",
      "|   null|      null|    null|  null|small_airport|Fazenda SÃ£o Fran...|        0|\n",
      "|   null|      null|    null|  null|small_airport|Presidente Vences...|        0|\n",
      "|   null|      null|    null|  null|small_airport|Fazenda das Repre...|        0|\n",
      "|   null|      null|    null|  null|small_airport|Fazenda SÃ£o JoÃ£...|        0|\n",
      "+-------+----------+--------+------+-------------+--------------------+---------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT f.FL_DATE, f.OP_CARRIER, f.TAIL_NUM,f.ORIGIN,\n",
    "                a.TYPE, a.NAME, a.IATA_CODE\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE \n",
    "                WHERE f.ORIGIN IS NULL\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "4894276a",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 165:>                                                        (0 + 3) / 3]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+\n",
      "|count(FL_DATE)|\n",
      "+--------------+\n",
      "|             0|\n",
      "+--------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 165:======================================>                  (2 + 1) / 3]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT COUNT(f.FL_DATE)\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE \n",
    "                WHERE f.ORIGIN IS NULL\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "id": "3cd1e4f1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 176:====>            (1 + 3) / 4][Stage 177:>                (0 + 1) / 2]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+--------+------+----+----+---------+\n",
      "|FL_DATE|OP_CARRIER|TAIL_NUM|ORIGIN|TYPE|NAME|IATA_CODE|\n",
      "+-------+----------+--------+------+----+----+---------+\n",
      "+-------+----------+--------+------+----+----+---------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 176:============>    (3 + 1) / 4][Stage 177:========>        (1 + 1) / 2]\r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT f.FL_DATE, f.OP_CARRIER, f.TAIL_NUM,f.ORIGIN,\n",
    "                a.TYPE, a.NAME, a.IATA_CODE\n",
    "                FROM flights AS f \n",
    "                RIGHT OUTER JOIN airportIATACodes AS a\n",
    "                ON f.ORIGIN = a.IATA_CODE \n",
    "                WHERE a.NAME IS NULL\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "b50b7fb6",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 182:======================================>                  (2 + 1) / 3]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+--------+------+\n",
      "|FL_DATE|OP_CARRIER|TAIL_NUM|ORIGIN|\n",
      "+-------+----------+--------+------+\n",
      "+-------+----------+--------+------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "spark.sql(\"\"\"SELECT FL_DATE,OP_CARRIER,TAIL_NUM,ORIGIN\n",
    "                    FROM flights\n",
    "                    WHERE ORIGIN IS NULL\"\"\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "2a489cdf",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+----------+--------+-----------------+-----------------+------+----------------+---------------+-----------+--------------+---------+---------+---------+--------+--------+--------------+\n",
      "|   FL_DATE|OP_CARRIER|TAIL_NUM|OP_CARRIER_FL_NUM|ORIGIN_AIRPORT_ID|ORIGIN|ORIGIN_CITY_NAME|DEST_AIRPORT_ID|DESTINATION|DEST_CITY_NAME|DEP_DELAY|ARR_DELAY|CANCELLED|AIR_TIME|DISTANCE|OCCUPANCY_RATE|\n",
      "+----------+----------+--------+-----------------+-----------------+------+----------------+---------------+-----------+--------------+---------+---------+---------+--------+--------+--------------+\n",
      "|2019-03-02|        WN|  N955WN|             4591|            14635|   RSW|  Fort Myers, FL|          11042|        CLE| Cleveland, OH|     -8.0|     -6.0|      0.0|   143.0|  1025.0|          0.97|\n",
      "|2019-03-02|        WN|  N8686A|             3231|            14635|   RSW|  Fort Myers, FL|          11066|        CMH|  Columbus, OH|      1.0|      5.0|      0.0|   135.0|   930.0|          0.55|\n",
      "+----------+----------+--------+-----------------+-----------------+------+----------------+---------------+-----------+--------------+---------+---------+---------+--------+--------+--------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM flights\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "b91ccbc1",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.window import Window\n",
    "spec = Window.partitionBy(\"FL_DATE\",\"ORIGIN\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "id": "d67e7aaf",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 17:>                                                         (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+----------+--------+---------+-----------+-----------+-----------+\n",
      "|FL_DATE|ORIGIN|OP_CARRIER|TAIL_NUM|DEP_DELAY|DepDelayMin|DepDelayMax|DepDelayAvg|\n",
      "+-------+------+----------+--------+---------+-----------+-----------+-----------+\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|         -5|         -5|       -5.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|         -2|         -2|       -2.0|\n",
      "+-------+------+----------+--------+---------+-----------+-----------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 17:=============================>                            (2 + 2) / 4]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "flights. \\\n",
    "    filter(\"CANCELLED = 0\"). \\\n",
    "    select(\"FL_DATE\",\"ORIGIN\",\"OP_CARRIER\",\"TAIL_NUM\",\n",
    "          col(\"DEP_DELAY\").cast(\"int\").alias(\"DEP_DELAY\")). \\\n",
    "    withColumn(\"DepDelayMin\",min(\"DEP_DELAY\").over(spec)). \\\n",
    "    withColumn(\"DepDelayMax\",max(\"DEP_DELAY\").over(spec)). \\\n",
    "    withColumn(\"DepDelayAvg\",avg(\"DEP_DELAY\").over(spec)). \\\n",
    "    orderBy(col(\"FL_DATE\"),col(\"ORIGIN\"),col(\"DEP_DELAY\").desc()).show(20)\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "2ac2374a",
   "metadata": {},
   "outputs": [],
   "source": [
    "rowBetween = Window.partitionBy(\"FL_DATE\", \"ORIGIN\") \\\n",
    "    .orderBy(\"DEP_DELAY\") \\\n",
    "    .rowsBetween(-3, Window.currentRow)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2a0adb88",
   "metadata": {},
   "source": [
    "We can also get cumulative aggregations based up on moving window. In this case we are getting cumulative aggregation using previous 3 records and current record.\n",
    "\n",
    "We can also use rangeBetween to get cumulative delay at each airport using scheduled departure time as sorting criteria. With Window.unboundedPreceding and Window.currentRow, the behavior is same as rowsBetween.\n",
    "\n",
    "rangeBetween considers the values rather than rows. With the logic below, it only consider those values which are between current row value and current row - 100 values based up on the CRSDepTime."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "id": "ab1cae98",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 20:>                                                         (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+----------+---------+-------------+\n",
      "|FL_DATE|ORIGIN|OP_CARRIER|DEP_DELAY|DEP_DELAY_SUM|\n",
      "+-------+------+----------+---------+-------------+\n",
      "| 1/1/19|   RSW|        B6|       -5|          -20|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -20|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -20|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -20|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -15|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -20|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -20|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -20|\n",
      "| 1/1/19|   RSW|        B6|       -5|           -5|\n",
      "| 1/1/19|   RSW|        B6|       -5|          -10|\n",
      "|1/12/19|   RSW|        DL|       -2|           -8|\n",
      "|1/12/19|   RSW|        DL|       -2|           -8|\n",
      "|1/12/19|   RSW|        DL|       -2|           -8|\n",
      "|1/12/19|   RSW|        DL|       -2|           -8|\n",
      "|1/12/19|   RSW|        DL|       -2|           -2|\n",
      "|1/12/19|   RSW|        DL|       -2|           -8|\n",
      "|1/12/19|   RSW|        DL|       -2|           -4|\n",
      "|1/12/19|   RSW|        DL|       -2|           -8|\n",
      "|1/12/19|   RSW|        DL|       -2|           -6|\n",
      "|1/12/19|   RSW|        DL|       -2|           -8|\n",
      "+-------+------+----------+---------+-------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "flights.filter(\"CANCELLED = 0\") \\\n",
    "    .select(\"FL_DATE\",\"ORIGIN\",\"OP_CARRIER\",\n",
    "            col(\"DEP_DELAY\").cast(\"int\").alias(\"DEP_DELAY\")) \\\n",
    "    .withColumn(\"DEP_DELAY_SUM\", sum(\"DEP_DELAY\").over(rowBetween)) \\\n",
    "    .orderBy(\"FL_DATE\",\"ORIGIN\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "210b4277",
   "metadata": {},
   "outputs": [],
   "source": [
    "unboundedSpec =  Window. \\\n",
    "    partitionBy(\"FL_DATE\", \"ORIGIN\"). \\\n",
    "    orderBy(\"DEP_DELAY\"). \\\n",
    "    rangeBetween(Window.unboundedPreceding, Window.currentRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "7da768fc",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 26:>                                                         (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+----------+--------+---------+-----------+\n",
      "|FL_DATE|ORIGIN|OP_CARRIER|TAIL_NUM|DEP_DELAY|DepDelaySum|\n",
      "+-------+------+----------+--------+---------+-----------+\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "+-------+------+----------+--------+---------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "flights. \\\n",
    "    filter(\"CANCELLED = 0\"). \\\n",
    "    select(\"FL_DATE\",\n",
    "           \"ORIGIN\",\n",
    "           \"OP_CARRIER\",\n",
    "           \"TAIL_NUM\",\n",
    "           col(\"DEP_DELAY\").cast(\"int\").alias(\"DEP_DELAY\")\n",
    "          ). \\\n",
    "    withColumn(\"DepDelaySum\", sum(\"DEP_DELAY\").over(unboundedSpec)). \\\n",
    "    orderBy(\"FL_DATE\", \"ORIGIN\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "bba98e73",
   "metadata": {},
   "outputs": [],
   "source": [
    "rangeBetweenSpec = Window. \\\n",
    "    partitionBy(\"FL_DATE\", \"ORIGIN\"). \\\n",
    "    orderBy(\"DEP_DELAY\"). \\\n",
    "    rangeBetween(-100, Window.currentRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "44ff60a3",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 29:=============================>                            (2 + 2) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+----------+--------+---------+-----------+\n",
      "|FL_DATE|ORIGIN|OP_CARRIER|TAIL_NUM|DEP_DELAY|DepDelaySum|\n",
      "+-------+------+----------+--------+---------+-----------+\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|        -50|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|        -20|\n",
      "+-------+------+----------+--------+---------+-----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 29:===========================================>              (3 + 1) / 4]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "flights. \\\n",
    "    filter(\"CANCELLED = 0\"). \\\n",
    "    select(\"FL_DATE\",\n",
    "           \"ORIGIN\",\n",
    "           \"OP_CARRIER\",\n",
    "           \"TAIL_NUM\",\n",
    "           col(\"DEP_DELAY\").cast(\"int\").alias(\"DEP_DELAY\")\n",
    "          ). \\\n",
    "    withColumn(\"DepDelaySum\", sum(\"DEP_DELAY\").over(rangeBetweenSpec)). \\\n",
    "    orderBy(\"FL_DATE\", \"ORIGIN\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "id": "985771f3",
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window. \\\n",
    "    partitionBy('FL_DATE',\"ORIGIN\"). \\\n",
    "    orderBy(col('DEP_DELAY').desc())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "id": "0cdaa6b1",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+----------+--------+---------+-----+-----+-----+---+\n",
      "|FL_DATE|ORIGIN|OP_CARRIER|TAIL_NUM|DEP_DELAY|srank|drank|prank| rn|\n",
      "+-------+------+----------+--------+---------+-----+-----+-----+---+\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  7|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  3|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  4|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0| 10|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  9|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  5|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  8|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  1|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  6|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|    1|    1|  0.0|  2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  8|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  5|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  6|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0| 10|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  9|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  7|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  1|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  4|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|    1|    1|  0.0|  3|\n",
      "+-------+------+----------+--------+---------+-----+-----+-----+---+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "flights. \\\n",
    "    filter(\"CANCELLED = 0\"). \\\n",
    "    select(\"FL_DATE\",\n",
    "           \"ORIGIN\",\n",
    "           \"OP_CARRIER\",\n",
    "           \"TAIL_NUM\",\n",
    "           col(\"DEP_DELAY\").cast(\"int\").alias(\"DEP_DELAY\")\n",
    "          ). \\\n",
    "    withColumn(\"srank\", rank().over(spec)). \\\n",
    "    withColumn(\"drank\", dense_rank().over(spec)). \\\n",
    "    withColumn(\"prank\", round(percent_rank().over(spec), 2)). \\\n",
    "    withColumn(\"rn\", row_number().over(spec)). \\\n",
    "    orderBy(\"FL_DATE\", \"ORIGIN\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "id": "08e25b32",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[Stage 41:>                                                         (0 + 4) / 4]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+----------+--------+---------+-----------------+-------------+--------------+\n",
      "|FL_DATE|ORIGIN|OP_CARRIER|TAIL_NUM|DEP_DELAY|LeadUniqueCarrier|LeadFlightNum|LeadCRSDepTime|\n",
      "+-------+------+----------+--------+---------+-----------------+-------------+--------------+\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|            -5|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|          null|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|             null|         null|          null|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|             null|       N965WN|          null|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|            -5|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|            -5|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|            -5|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|            -5|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|            -5|\n",
      "| 1/1/19|   RSW|        B6|  N965WN|       -5|               B6|       N965WN|            -5|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|            -2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|            -2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|          null|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|             null|         null|          null|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|            -2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|            -2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|            -2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|             null|       N8317M|          null|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|            -2|\n",
      "|1/12/19|   RSW|        DL|  N8317M|       -2|               DL|       N8317M|            -2|\n",
      "+-------+------+----------+--------+---------+-----------------+-------------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 41:=============================>                            (2 + 2) / 4]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "flights. \\\n",
    "    filter(\"CANCELLED = 0\"). \\\n",
    "    select(\"FL_DATE\",\n",
    "           \"ORIGIN\",\n",
    "           \"OP_CARRIER\",\n",
    "           \"TAIL_NUM\",\n",
    "           col(\"DEP_DELAY\").cast(\"int\").alias(\"DEP_DELAY\")\n",
    "          ). \\\n",
    "    withColumn(\"LeadUniqueCarrier\", lead(\"OP_CARRIER\",2).over(spec)). \\\n",
    "    withColumn(\"LeadFlightNum\", lead(\"TAIL_NUM\").over(spec)). \\\n",
    "    withColumn(\"LeadCRSDepTime\", lead(\"DEP_DELAY\",3).over(spec)). \\\n",
    "    orderBy(\"FL_DATE\", \"ORIGIN\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9a82fd1b",
   "metadata": {},
   "source": [
    "Let us get highest paid employee and least paid employee with in each department for each employee using employees data set.\n",
    "\n",
    "\n",
    "You can also use max to get max salary for each department, but you cannot get other attributes related to ma salary such as employee id, name etc. With first or last, you can get other details as well."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f7063991",
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window. \\\n",
    "    partitionBy('department_id'). \\\n",
    "    orderBy(col('salary').desc())\n",
    "\n",
    "employees. \\\n",
    "    select('employee_id', \n",
    "           col('department_id').cast('int').alias('department_id'), \n",
    "           'salary'\n",
    "          ). \\\n",
    "    withColumn(\"highest_salary\", first('salary').over(spec)). \\\n",
    "    withColumn(\"highest_employee_id\", first('employee_id').over(spec)). \\\n",
    "    orderBy(\"department_id\", col(\"salary\").desc()). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "99ba9daf",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "803f8ad5",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "86ed4827",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f4e616cf",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "95b9a54f",
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "09e72261",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}