07_windowing_functions/03_aggregate_functions.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Aggregate Functions\n",
    "\n",
    "Let us see how to perform aggregations within each group while projecting the raw data that is used to perform the aggregation.\n",
    "\n",
    " * We have functions such as `sum`, `avg`, `min`, `max` etc which can be used to aggregate the data.\n",
    " * We need to create `WindowSpec` object using `partitionBy` to get aggregations within each group.\n",
    " * Typically we don’t need to sort the data to perform aggregations, however if we want to perform cumulative aggregations using rowsBetween, then we have to sort the data using cumulative criteria.\n",
    " * Let us try to get total departure delay, minimum departure delay, maximum departure delay and average departure delay for each day for each airport. We will ignore all those flights which are departed early or ontime."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let us start spark context for this Notebook so that we can execute the code provided. You can sign up for our [10 node state of the art cluster/labs](https://labs.itversity.com/plans) to learn Spark SQL using our unique integrated LMS."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "import getpass\n",
    "username = getpass.getuser()\n",
    "\n",
    "spark = SparkSession. \\\n",
    "    builder. \\\n",
    "    config('spark.ui.port', '0'). \\\n",
    "    config(\"spark.sql.warehouse.dir\", f\"/user/{username}/warehouse\"). \\\n",
    "    enableHiveSupport(). \\\n",
    "    appName(f'{username} | Python - Windowing Functions'). \\\n",
    "    master('yarn'). \\\n",
    "    getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "If you are going to use CLIs, you can use Spark SQL using one of the 3 approaches.\n",
    "\n",
    "**Using Spark SQL**\n",
    "\n",
    "```\n",
    "spark2-sql \\\n",
    "    --master yarn \\\n",
    "    --conf spark.ui.port=0 \\\n",
    "    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse\n",
    "```\n",
    "\n",
    "**Using Scala**\n",
    "\n",
    "```\n",
    "spark2-shell \\\n",
    "    --master yarn \\\n",
    "    --conf spark.ui.port=0 \\\n",
    "    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse\n",
    "```\n",
    "\n",
    "**Using Pyspark**\n",
    "\n",
    "```\n",
    "pyspark2 \\\n",
    "    --master yarn \\\n",
    "    --conf spark.ui.port=0 \\\n",
    "    --conf spark.sql.warehouse.dir=/user/${USER}/warehouse\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.conf.set('spark.sql.shuffle.partitions', '2')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "airtraffic_path = \"/public/airtraffic_all/airtraffic-part/flightmonth=200801\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "airtraffic = spark. \\\n",
    "  read. \\\n",
    "  parquet(airtraffic_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Year: integer (nullable = true)\n",
      " |-- Month: integer (nullable = true)\n",
      " |-- DayofMonth: integer (nullable = true)\n",
      " |-- DayOfWeek: integer (nullable = true)\n",
      " |-- DepTime: string (nullable = true)\n",
      " |-- CRSDepTime: integer (nullable = true)\n",
      " |-- ArrTime: string (nullable = true)\n",
      " |-- CRSArrTime: integer (nullable = true)\n",
      " |-- UniqueCarrier: string (nullable = true)\n",
      " |-- FlightNum: integer (nullable = true)\n",
      " |-- TailNum: string (nullable = true)\n",
      " |-- ActualElapsedTime: string (nullable = true)\n",
      " |-- CRSElapsedTime: integer (nullable = true)\n",
      " |-- AirTime: string (nullable = true)\n",
      " |-- ArrDelay: string (nullable = true)\n",
      " |-- DepDelay: string (nullable = true)\n",
      " |-- Origin: string (nullable = true)\n",
      " |-- Dest: string (nullable = true)\n",
      " |-- Distance: string (nullable = true)\n",
      " |-- TaxiIn: string (nullable = true)\n",
      " |-- TaxiOut: string (nullable = true)\n",
      " |-- Cancelled: integer (nullable = true)\n",
      " |-- CancellationCode: string (nullable = true)\n",
      " |-- Diverted: integer (nullable = true)\n",
      " |-- CarrierDelay: string (nullable = true)\n",
      " |-- WeatherDelay: string (nullable = true)\n",
      " |-- NASDelay: string (nullable = true)\n",
      " |-- SecurityDelay: string (nullable = true)\n",
      " |-- LateAircraftDelay: string (nullable = true)\n",
      " |-- IsArrDelayed: string (nullable = true)\n",
      " |-- IsDepDelayed: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airtraffic.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import col, lit, lpad, concat"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import min, max, sum, avg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.window import Window"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on function partitionBy in module pyspark.sql.window:\n",
      "\n",
      "partitionBy(*cols)\n",
      "    Creates a :class:`WindowSpec` with the partitioning defined.\n",
      "    \n",
      "    .. versionadded:: 1.4\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(Window.partitionBy)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window. \\\n",
    "    partitionBy(\"FlightDate\", \"Origin\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.window.WindowSpec"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(spec)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+---------+----------+------------+--------+-----------+-----------+-----------+------------------+\n",
      "|FlightDate|Origin|UniqueCarrier|FlightNum|CRSDepTime|IsDepDelayed|DepDelay|DepDelayMin|DepDelayMax|DepDelaySum|       DepDelayAvg|\n",
      "+----------+------+-------------+---------+----------+------------+--------+-----------+-----------+-----------+------------------+\n",
      "|  20080101|   ABE|           OO|     5873|       720|         YES|       1|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABE|           OH|     5457|      1720|         YES|      14|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABE|           XE|     2578|      1410|         YES|      22|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABE|           9E|     2936|      1615|         YES|      34|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABE|           XE|     2594|      1740|         YES|      34|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABE|           9E|     2940|      1215|         YES|      70|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABE|           YV|     7263|      1230|         YES|     137|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABE|           YV|     7138|      1741|         YES|     175|          1|        175|        487|            60.875|\n",
      "|  20080101|   ABI|           MQ|     3214|      1735|         YES|       3|          3|          3|          3|               3.0|\n",
      "|  20080101|   ABQ|           WN|     2976|      1040|         YES|       1|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|      972|      1810|         YES|       1|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|       61|      1320|         YES|       1|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|     3425|      1440|         YES|       2|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|       88|       755|         YES|       2|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|     2284|      1520|         YES|       3|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           XE|     2771|      1430|         YES|       3|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|     1493|      2020|         YES|       4|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|      360|      1800|         YES|       5|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|      644|      1725|         YES|       5|          1|        218|       1580|32.916666666666664|\n",
      "|  20080101|   ABQ|           WN|      729|      1530|         YES|       7|          1|        218|       1580|32.916666666666664|\n",
      "+----------+------+-------------+---------+----------+------------+--------+-----------+-----------+-----------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "airtraffic. \\\n",
    "    filter(\"IsDepDelayed = 'YES' and Cancelled = 0\"). \\\n",
    "    select(concat(\"Year\", \n",
    "                  lpad(\"Month\", 2, \"0\"), \n",
    "                  lpad(\"DayOfMonth\", 2, \"0\")\n",
    "                 ).alias(\"FlightDate\"),\n",
    "           \"Origin\",\n",
    "           \"UniqueCarrier\",\n",
    "           \"FlightNum\",\n",
    "           \"CRSDepTime\",\n",
    "           \"IsDepDelayed\",\n",
    "           col(\"DepDelay\").cast(\"int\").alias(\"DepDelay\")\n",
    "          ). \\\n",
    "    withColumn(\"DepDelayMin\", min(\"DepDelay\").over(spec)). \\\n",
    "    withColumn(\"DepDelayMax\", max(\"DepDelay\").over(spec)). \\\n",
    "    withColumn(\"DepDelaySum\", sum(\"DepDelay\").over(spec)). \\\n",
    "    withColumn(\"DepDelayAvg\", avg(\"DepDelay\").over(spec)). \\\n",
    "    orderBy(\"FlightDate\", \"Origin\", \"DepDelay\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Pyspark 2",
   "language": "python",
   "name": "pyspark2"
  },
  "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.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}