spark/Spark-SQL-queries.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from datetime import datetime\n",
    "from pyspark.sql.types import IntegerType, FloatType\n",
    "from pyspark.sql.functions import udf,lit\n",
    "\n",
    "from pyspark.sql import SQLContext\n",
    "\n",
    "sc = pyspark.SparkContext.getOrCreate()\n",
    "sqlContext = SQLContext(sc)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Queries:\n",
    "1. the percentage of canceled flights per day, throughout the entire data set\n",
    "2. weekly percentages of delays that are due to weather, throughout the entire data set \n",
    "3. the percentage of flights belonging to a given \"distance group\" that were able to halve their departure delays by the time they arrived at their destinations. Distance groups assort flights by their total distance in miles. Flights with distances that are less than 200 miles belong in group 1, flights with distances that are between 200 and 399 miles belong in group 2, flights with distances that are between 400 and 599 miles belong in group 3, and so on. The last group contains flights whose distances are between 2400 and 2599 miles.\n",
    "4. a weekly \"penalty\" score for each airport that depends on both the its incoming and outgoing flights. The score adds 0.5 for each incoming flight that is more than 15 minutes late, and 1 for each outgoing flight that is more than 15 minutes late.\n",
    "\n",
    "\n",
    "## 1. \"The percentage of canceled flights per day, throughout the entire data set\"\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getCancelledPercentage(file):\n",
    "    d = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "        .load(file)\n",
    "    \n",
    "    d_ymdc= d.select(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"],d[\"Cancelled\"])\n",
    "    \n",
    "    d_cancelled = d_ymdc.filter(d[\"Cancelled\"]==1) \\\n",
    "        .groupBy(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"]) \\\n",
    "        .count().withColumnRenamed(\"count\",\"cancelled\")\n",
    "    \n",
    "    d_total = d_ymdc \\\n",
    "        .groupBy(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"]) \\\n",
    "        .count().withColumnRenamed(\"count\",\"total\")\n",
    "    \n",
    "    res = d_cancelled.join(d_total, on=[d_cancelled.Year == d_total.Year, d_cancelled.Month == d_total.Month,d_cancelled.DayofMonth == d_total.DayofMonth]) \\\n",
    "        .select(d_cancelled.Year,d_cancelled.Month,d_cancelled.DayofMonth,d_cancelled[\"cancelled\"],d_total[\"total\"])\n",
    "    \n",
    "    return res.withColumn(\"percentageCancelled\", (res.cancelled/res.total))\\\n",
    "        .drop(\"cancelled\",\"total\")\\\n",
    "        .orderBy(\"Year\",\"Month\",\"DayofMonth\")\n",
    "        \n",
    "\n",
    "res1=[getCancelledPercentage(\"../BDdata/\"+str(i)+\".csv\") for i in range(1994,2009)]\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "def saveUnited(filename, res1):\n",
    "    res=res1[0]\n",
    "    for el in res1[1:]:\n",
    "        res=res.union(el)\n",
    "    res.coalesce(1).write.csv(filename,header=True)\n",
    "\n",
    "saveUnited(\"./results/query1.csv\",res1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Extra query: Weekly percentages of flights cancelled that are due to weather, throughout the entire data set "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def getWeatherPercentage(year, newYearFirstWeek=None):\n",
    "    d = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "        .load('../BDdata/'+year+'.csv')\n",
    "    dr = d.filter(d[\"Cancelled\"]=='1')\\\n",
    "        .select(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"],d[\"CancellationCode\"])\n",
    "\n",
    "    d_cancelled_weather = dr.filter(d[\"CancellationCode\"]=='B')\n",
    "\n",
    "    d2 = dr.groupBy(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"])\\\n",
    "        .count()\\\n",
    "        .withColumnRenamed(\"count\",\"cancelled\")\n",
    "    \n",
    "    d3 = d_cancelled_weather.groupBy(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"])\\\n",
    "        .count()\\\n",
    "        .withColumnRenamed(\"count\",\"weather\")\n",
    "    \n",
    "    d4 = d2.join(d3, on=[d2[\"Year\"]==d3[\"Year\"],d2[\"Month\"]==d3[\"Month\"],d2[\"DayofMonth\"]==d3[\"DayofMonth\"]])\\\n",
    "        .select(d2[\"Year\"],d2[\"Month\"],d2[\"DayofMonth\"],d2[\"cancelled\"],d3[\"weather\"])\\\n",
    "        .orderBy(d2[\"Year\"],d2[\"Month\"],d2[\"DayofMonth\"])\n",
    "        \n",
    "    d5 = d4.withColumn(\"week\", week(d4[\"Year\"],d4[\"Month\"],d4[\"DayofMonth\"]))\n",
    "    \n",
    "    newYearFirstWeek1 = d5.filter(d5[\"week\"]==1).filter(d5[\"Month\"]==\"12\")\n",
    "    if newYearFirstWeek is not None:\n",
    "        d5 = d5.union(newYearFirstWeek)\n",
    "    d5a = d5.filter((d5[\"week\"]!=1) | (d5[\"Month\"]!=\"12\")).groupBy(\"Year\",\"week\")\\\n",
    "        .agg({\"cancelled\":\"sum\",\"weather\":\"sum\"})\n",
    "    d6 = d5a.withColumn(\"percentage\",d5a[\"sum(weather)\"]/d5a[\"sum(cancelled)\"])\\\n",
    "        .select(d5a[\"Year\"],d5a[\"week\"],\"percentage\")\\\n",
    "        .sort(\"Year\",\"week\")\n",
    "    return (d6, newYearFirstWeek1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Weekly percentages of delays that are due to weather, throughout the entire data set "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def getWeatherDelayPercentage(start, end):\n",
    "    d = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "        .load('../BDdata/'+str(start)+'.csv')\n",
    "    for i in range(start+1,end+1):\n",
    "        d = d.union(sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "                .load('../BDdata/'+str(i)+'.csv'))\n",
    "    d1 = d.where(\"ArrDelay >= 0\").select(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"],d[\"ArrDelay\"],d[\"WeatherDelay\"])\n",
    "    # needed check on presence of \"NA\" because otherwise an empty table is returned if filtered when no \"NA\" is present\n",
    "    if d1.where(\"WeatherDelay == 'NA'\").count() > 0:\n",
    "        d1 = d1.where(\"WeatherDelay != 'NA'\")\n",
    "\n",
    "    d3 = d1.groupBy(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"]).agg({\"ArrDelay\":\"sum\",\"WeatherDelay\":\"sum\"})\n",
    "    \n",
    "    week = udf(lambda *x : datetime(x[0],x[1],x[2]).isocalendar()[1], IntegerType())\n",
    "    year = udf(lambda *x : datetime(x[0],x[1],x[2]).isocalendar()[0], IntegerType())\n",
    "\n",
    "    d4 = d3.withColumn(\"Week\", week(d3[\"Year\"],d3[\"Month\"],d3[\"DayofMonth\"]))\\\n",
    "        .withColumn(\"Year\",year(d3[\"Year\"],d3[\"Month\"],d3[\"DayofMonth\"])).drop(d3[\"Year\"])\n",
    "\n",
    "    d5 = d4.groupBy(\"Year\",\"Week\")\\\n",
    "            .agg({\"sum(WeatherDelay)\":\"sum\",\"sum(ArrDelay)\":\"sum\"})\n",
    "    d6 = d5.withColumn(\"percentage\",d5[\"sum(sum(WeatherDelay))\"]/d5[\"sum(sum(ArrDelay))\"])\\\n",
    "            .drop(\"sum(sum(WeatherDelay))\",\"sum(sum(ArrDelay))\")\\\n",
    "            .sort(\"Year\",\"Week\")\n",
    "    \n",
    "    return d6\n",
    "            \n",
    "d7 = getWeatherDelayPercentage(1994,2008)\n",
    "d7.coalesce(1).write.csv(\"./results/query2.csv\",header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. The percentage of flights belonging to a given \"distance group\" that were able to halve their departure delays by the time they arrived at their destinations. \n",
    "\n",
    "Distance groups assort flights by their total distance in miles. Flights with distances that are less than 200 miles belong in group 1, flights with distances that are between 200 and 399 miles belong in group 2, flights with distances that are between 400 and 599 miles belong in group 3, and so on. The last group contains flights whose distances are between 2400 and 2599 miles."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def getPercentageFastDistanceGroup(init,end):\n",
    "    d = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "        .load('../BDdata/'+str(init)+'.csv')\n",
    "    for i in range(init+1,end+1):\n",
    "        d = d.union(sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "            .load('../BDdata/'+str(i)+'.csv'))\n",
    "\n",
    "    d1 = d.where(d[\"Cancelled\"]==0)\\\n",
    "        .where(d[\"DepDelay\"] != \"NA\")\\\n",
    "        .where(d[\"ArrDelay\"] != \"NA\")\\\n",
    "        .where(d[\"Distance\"] != \"NA\")\\\n",
    "        .select(d[\"DepDelay\"],d[\"ArrDelay\"],d[\"Distance\"])\n",
    "    \n",
    "    distanceGroup = udf(lambda x : int(x)//200+1,IntegerType())\n",
    "    d2 = d1.withColumn(\"DistanceGroup\",distanceGroup(d1[\"Distance\"]))\\\n",
    "        .drop(d1[\"Distance\"])\n",
    "    \n",
    "    d3 = d2.where(\"DepDelay > 0\")\\\n",
    "        .filter(d2[\"DepDelay\"]>=2*d2[\"ArrDelay\"])\\\n",
    "        .groupBy(d2.DistanceGroup).agg({\"*\":\"count\"})\n",
    "\n",
    "    d4 = d2.groupBy(d2.DistanceGroup).agg({\"*\":\"count\"})\n",
    "    \n",
    "    d5 = d4.join(d3,on=[d3.DistanceGroup==d4.DistanceGroup])\\\n",
    "        .withColumn(\"percentage\", d3[\"count(1)\"]/d4[\"count(1)\"])\\\n",
    "        .drop(\"count(1)\")\\\n",
    "        .select(d4.DistanceGroup,\"percentage\").sort(\"DistanceGroup\")\n",
    "    return d5\n",
    "\n",
    "res4 = getPercentageFastDistanceGroup(1994,2008)\n",
    "res4.coalesce(1).write.csv(\"./results/query3.csv\",header=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark.sql.functions as sf\n",
    "\n",
    "def getPercentageFastDistanceGroup(init,end):\n",
    "    d = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "        .load('../BDdata/'+str(init)+'.csv')\n",
    "    for i in range(init+1,end+1):\n",
    "        d = d.union(sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "            .load('../BDdata/'+str(i)+'.csv'))\n",
    "\n",
    "    d1 = d.where(d[\"Cancelled\"]==0)\\\n",
    "        .where(d[\"DepDelay\"] != \"NA\")\\\n",
    "        .where(d[\"ArrDelay\"] != \"NA\")\\\n",
    "        .where(d[\"Distance\"] != \"NA\")\n",
    "    d1 = d1.select(d[\"Origin\"],d[\"Dest\"],d[\"Distance\"])\n",
    "    \n",
    "    distanceGroup = udf(lambda x : int(x)//200+1,IntegerType())\n",
    "    d2 = d1.withColumn(\"DistanceGroup\",distanceGroup(d1[\"Distance\"]))\\\n",
    "        .drop(d1[\"Distance\"])\n",
    "    \n",
    "    d3 = d2.groupBy(d2.Origin, d2.Dest, d2.DistanceGroup).agg(sf.count(\"*\").alias(\"Num\"))\n",
    "    d4a = d3.select(d3.Origin, d2.DistanceGroup, d3.Num).groupBy(d3.Origin,d3.DistanceGroup).agg(sf.sum(d3.Num).alias(\"Out\"))\n",
    "    d4b = d3.select(d3.Dest, d2.DistanceGroup, d3.Num).groupBy(d3.Dest,d3.DistanceGroup).agg(sf.sum(d3.Num).alias(\"In\"))\n",
    "    d5 = d4a.join(d4b, on=[d4a.Origin == d4b.Dest, d4a.DistanceGroup == d4b.DistanceGroup])\\\n",
    "        .select(d4a.DistanceGroup, d4a.Origin, d4a.Out, d4b.In)\n",
    "    d5 = d5.withColumn(\"Total\", d5.Out + d5.In)\\\n",
    "        .orderBy(\"DistanceGroup\", \"Origin\")\n",
    "        \n",
    "    return d5\n",
    "\n",
    "res4 = getPercentageFastDistanceGroup(1994,2008)\n",
    "res4.coalesce(1).write.csv(\"./results/query3.csv/extra\",header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. A weekly \"penalty\" score for each airport that depends on both the its incoming and outgoing flights. \n",
    "The score adds 0.5 for each incoming flight that is more than 15 minutes late, and 1 for each outgoing flight that is more than 15 minutes late."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "week = udf(lambda *x : datetime(x[0],x[1],x[2]).isocalendar()[1], IntegerType())\n",
    "year = udf(lambda *x : datetime(x[0],x[1],x[2]).isocalendar()[0], IntegerType())\n",
    "penalty = lambda y : udf(lambda x: y if int(x) > 15 else 0.0, FloatType())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "\n",
    "def getAirportPenalty(start, end):\n",
    "    \n",
    "    d = sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "        .load('../BDdata/'+str(start)+'.csv')\n",
    "    for i in range(start+1,end+1):\n",
    "        d = d.union(sqlContext.read.format('com.databricks.spark.csv').options(header='true', inferschema='true')\\\n",
    "                .load('../BDdata/'+str(i)+'.csv'))\n",
    "    d1 = d.withColumn(\"Week\", week(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"]))\\\n",
    "        .withColumn(\"Year\", year(d[\"Year\"],d[\"Month\"],d[\"DayofMonth\"])).drop(d[\"Year\"])\n",
    "    d1  = d1.filter((d1['DepDelay']!= \"NA\") & (d1['ArrDelay'] != \"NA\") & ((d1['ArrDelay'] > 15) | (d1['DepDelay'] > 15)))\n",
    "    d2 = d1.withColumn(\"in\", penalty(0.5)(d1['ArrDelay']))\\\n",
    "        .withColumn(\"out\", penalty(1.0)(d1['DepDelay']))\\\n",
    "        .select(d1[\"Year\"],d1[\"Week\"],d1[\"Origin\"],d1[\"Dest\"],\"in\",\"out\")\n",
    "    d3 = d2.groupBy(d2[\"Year\"],d2[\"Origin\"],d2[\"Week\"])\\\n",
    "        .agg({\"out\":\"sum\"})\n",
    "    d3a = d2.groupBy(d2[\"Year\"],d2[\"Dest\"],d2[\"Week\"]).agg({\"in\":\"sum\"})\n",
    "    d3b = d3a.join(d3, on=[d3a[\"Dest\"]==d3[\"Origin\"],d3a[\"Year\"]==d3[\"Year\"],d3a[\"Week\"]==d3[\"Week\"]])\\\n",
    "            .select(d3[\"Year\"],d3[\"Week\"],d3[\"sum(out)\"],d3a[\"sum(in)\"],d3[\"Origin\"])\n",
    "    \n",
    "    d4 = d3b.withColumn(\"Penalty\", d3b[\"sum(out)\"]+d3b[\"sum(in)\"])\\\n",
    "        .drop(\"Dest\")\\\n",
    "        .withColumnRenamed(\"Origin\",\"Airport\")\\\n",
    "        .withColumnRenamed(\"sum(out)\",\"OutgoingPenalty\")\\\n",
    "        .withColumnRenamed(\"sum(in)\",\"IngoingPenalty\")\\\n",
    "        .sort(\"Penalty\")\n",
    "    return d4\n",
    "\n",
    "res5 = getAirportPenalty(1994,2008)\n",
    "res5.coalesce(1).write.csv(\"./results/query4.csv\",header=True)"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.6"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}