07_windowing_functions/04_using_rowsBetween_and_rangeBetween.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Using rowsBetween and rangeBetween\n",
    "\n",
    "We can get cumulative aggregations using `rowsBetween` or `rangeBetween`.\n",
    "\n",
    "* We can use `rowsBetween` to include particular set of rows to perform aggregations.\n",
    "* We can use `rangeBetween` to include particular range of values on a given column."
   ]
  },
  {
   "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": [],
   "source": [
    "from pyspark.sql.window import Window"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Let us get cumulative delay at each airport using scheduled departure time as sorting criteria."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window. \\\n",
    "    partitionBy(\"FlightDate\", \"Origin\"). \\\n",
    "    orderBy(\"CRSDepTime\"). \\\n",
    "    rowsBetween(Window.unboundedPreceding, Window.currentRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import lpad, concat, col, sum"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|FlightDate|Origin|UniqueCarrier|FlightNum|CRSDepTime|IsDepDelayed|DepDelay|DepDelaySum|\n",
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|  20080101|   ABE|           OO|     5873|       720|         YES|       1|          1|\n",
      "|  20080101|   ABE|           9E|     2940|      1215|         YES|      70|         71|\n",
      "|  20080101|   ABE|           YV|     7263|      1230|         YES|     137|        208|\n",
      "|  20080101|   ABE|           XE|     2578|      1410|         YES|      22|        230|\n",
      "|  20080101|   ABE|           9E|     2936|      1615|         YES|      34|        264|\n",
      "|  20080101|   ABE|           OH|     5457|      1720|         YES|      14|        278|\n",
      "|  20080101|   ABE|           XE|     2594|      1740|         YES|      34|        312|\n",
      "|  20080101|   ABE|           YV|     7138|      1741|         YES|     175|        487|\n",
      "|  20080101|   ABI|           MQ|     3214|      1735|         YES|       3|          3|\n",
      "|  20080101|   ABQ|           DL|     1601|       740|         YES|      65|         65|\n",
      "|  20080101|   ABQ|           WN|       88|       755|         YES|       2|         67|\n",
      "|  20080101|   ABQ|           UA|      782|       805|         YES|      48|        115|\n",
      "|  20080101|   ABQ|           AA|     1814|       925|         YES|     100|        215|\n",
      "|  20080101|   ABQ|           WN|     2976|      1040|         YES|       1|        216|\n",
      "|  20080101|   ABQ|           WN|     1185|      1140|         YES|       9|        225|\n",
      "|  20080101|   ABQ|           WN|     1297|      1155|         YES|      12|        237|\n",
      "|  20080101|   ABQ|           WN|     2480|      1200|         YES|      16|        253|\n",
      "|  20080101|   ABQ|           WN|      137|      1220|         YES|      10|        263|\n",
      "|  20080101|   ABQ|           WN|     1328|      1245|         YES|      78|        341|\n",
      "|  20080101|   ABQ|           WN|     3245|      1250|         YES|      24|        365|\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(\"DepDelaySum\", sum(\"DepDelay\").over(spec)). \\\n",
    "    orderBy(\"FlightDate\", \"Origin\", \"CRSDepTime\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "markdown",
   "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."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window. \\\n",
    "    partitionBy(\"FlightDate\", \"Origin\"). \\\n",
    "    orderBy(\"CRSDepTime\"). \\\n",
    "    rowsBetween(-3, Window.currentRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|FlightDate|Origin|UniqueCarrier|FlightNum|CRSDepTime|IsDepDelayed|DepDelay|DepDelaySum|\n",
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|  20080101|   ABE|           OO|     5873|       720|         YES|       1|          1|\n",
      "|  20080101|   ABE|           9E|     2940|      1215|         YES|      70|         71|\n",
      "|  20080101|   ABE|           YV|     7263|      1230|         YES|     137|        208|\n",
      "|  20080101|   ABE|           XE|     2578|      1410|         YES|      22|        230|\n",
      "|  20080101|   ABE|           9E|     2936|      1615|         YES|      34|        263|\n",
      "|  20080101|   ABE|           OH|     5457|      1720|         YES|      14|        207|\n",
      "|  20080101|   ABE|           XE|     2594|      1740|         YES|      34|        104|\n",
      "|  20080101|   ABE|           YV|     7138|      1741|         YES|     175|        257|\n",
      "|  20080101|   ABI|           MQ|     3214|      1735|         YES|       3|          3|\n",
      "|  20080101|   ABQ|           DL|     1601|       740|         YES|      65|         65|\n",
      "|  20080101|   ABQ|           WN|       88|       755|         YES|       2|         67|\n",
      "|  20080101|   ABQ|           UA|      782|       805|         YES|      48|        115|\n",
      "|  20080101|   ABQ|           AA|     1814|       925|         YES|     100|        215|\n",
      "|  20080101|   ABQ|           WN|     2976|      1040|         YES|       1|        151|\n",
      "|  20080101|   ABQ|           WN|     1185|      1140|         YES|       9|        158|\n",
      "|  20080101|   ABQ|           WN|     1297|      1155|         YES|      12|        122|\n",
      "|  20080101|   ABQ|           WN|     2480|      1200|         YES|      16|         38|\n",
      "|  20080101|   ABQ|           WN|      137|      1220|         YES|      10|         47|\n",
      "|  20080101|   ABQ|           WN|     1328|      1245|         YES|      78|        116|\n",
      "|  20080101|   ABQ|           WN|     3245|      1250|         YES|      24|        128|\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(\"DepDelaySum\", sum(\"DepDelay\").over(spec)). \\\n",
    "    orderBy(\"FlightDate\", \"Origin\", \"CRSDepTime\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* 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`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window. \\\n",
    "    partitionBy(\"FlightDate\", \"Origin\"). \\\n",
    "    orderBy(\"CRSDepTime\"). \\\n",
    "    rangeBetween(Window.unboundedPreceding, Window.currentRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import lpad, concat, col, sum"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|FlightDate|Origin|UniqueCarrier|FlightNum|CRSDepTime|IsDepDelayed|DepDelay|DepDelaySum|\n",
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|  20080101|   ABE|           OO|     5873|       720|         YES|       1|          1|\n",
      "|  20080101|   ABE|           9E|     2940|      1215|         YES|      70|         71|\n",
      "|  20080101|   ABE|           YV|     7263|      1230|         YES|     137|        208|\n",
      "|  20080101|   ABE|           XE|     2578|      1410|         YES|      22|        230|\n",
      "|  20080101|   ABE|           9E|     2936|      1615|         YES|      34|        264|\n",
      "|  20080101|   ABE|           OH|     5457|      1720|         YES|      14|        278|\n",
      "|  20080101|   ABE|           XE|     2594|      1740|         YES|      34|        312|\n",
      "|  20080101|   ABE|           YV|     7138|      1741|         YES|     175|        487|\n",
      "|  20080101|   ABI|           MQ|     3214|      1735|         YES|       3|          3|\n",
      "|  20080101|   ABQ|           DL|     1601|       740|         YES|      65|         65|\n",
      "|  20080101|   ABQ|           WN|       88|       755|         YES|       2|         67|\n",
      "|  20080101|   ABQ|           UA|      782|       805|         YES|      48|        115|\n",
      "|  20080101|   ABQ|           AA|     1814|       925|         YES|     100|        215|\n",
      "|  20080101|   ABQ|           WN|     2976|      1040|         YES|       1|        216|\n",
      "|  20080101|   ABQ|           WN|     1185|      1140|         YES|       9|        225|\n",
      "|  20080101|   ABQ|           WN|     1297|      1155|         YES|      12|        237|\n",
      "|  20080101|   ABQ|           WN|     2480|      1200|         YES|      16|        253|\n",
      "|  20080101|   ABQ|           WN|      137|      1220|         YES|      10|        263|\n",
      "|  20080101|   ABQ|           WN|     1328|      1245|         YES|      78|        341|\n",
      "|  20080101|   ABQ|           WN|     3245|      1250|         YES|      24|        365|\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(\"DepDelaySum\", sum(\"DepDelay\").over(spec)). \\\n",
    "    orderBy(\"FlightDate\", \"Origin\", \"CRSDepTime\"). \\\n",
    "    show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* `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": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window. \\\n",
    "    partitionBy(\"FlightDate\", \"Origin\"). \\\n",
    "    orderBy(\"CRSDepTime\"). \\\n",
    "    rangeBetween(-100, Window.currentRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|FlightDate|Origin|UniqueCarrier|FlightNum|CRSDepTime|IsDepDelayed|DepDelay|DepDelaySum|\n",
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "|  20080101|   ABE|           OO|     5873|       720|         YES|       1|          1|\n",
      "|  20080101|   ABE|           9E|     2940|      1215|         YES|      70|         70|\n",
      "|  20080101|   ABE|           YV|     7263|      1230|         YES|     137|        207|\n",
      "|  20080101|   ABE|           XE|     2578|      1410|         YES|      22|         22|\n",
      "|  20080101|   ABE|           9E|     2936|      1615|         YES|      34|         34|\n",
      "|  20080101|   ABE|           OH|     5457|      1720|         YES|      14|         14|\n",
      "|  20080101|   ABE|           XE|     2594|      1740|         YES|      34|         48|\n",
      "|  20080101|   ABE|           YV|     7138|      1741|         YES|     175|        223|\n",
      "|  20080101|   ABI|           MQ|     3214|      1735|         YES|       3|          3|\n",
      "|  20080101|   ABQ|           DL|     1601|       740|         YES|      65|         65|\n",
      "|  20080101|   ABQ|           WN|       88|       755|         YES|       2|         67|\n",
      "|  20080101|   ABQ|           UA|      782|       805|         YES|      48|        115|\n",
      "|  20080101|   ABQ|           AA|     1814|       925|         YES|     100|        100|\n",
      "|  20080101|   ABQ|           WN|     2976|      1040|         YES|       1|          1|\n",
      "|  20080101|   ABQ|           WN|     1185|      1140|         YES|       9|         10|\n",
      "|  20080101|   ABQ|           WN|     1297|      1155|         YES|      12|         21|\n",
      "|  20080101|   ABQ|           WN|     2480|      1200|         YES|      16|         37|\n",
      "|  20080101|   ABQ|           WN|      137|      1220|         YES|      10|         47|\n",
      "|  20080101|   ABQ|           WN|     1328|      1245|         YES|      78|        116|\n",
      "|  20080101|   ABQ|           WN|     3245|      1250|         YES|      24|        140|\n",
      "|  20080101|   ABQ|           WN|       61|      1320|         YES|       1|        113|\n",
      "|  20080101|   ABQ|           DL|      940|      1345|         YES|       9|        112|\n",
      "|  20080101|   ABQ|           WN|      734|      1400|         YES|      42|         52|\n",
      "|  20080101|   ABQ|           XE|     2771|      1430|         YES|       3|         54|\n",
      "|  20080101|   ABQ|           WN|       45|      1435|         YES|      64|        118|\n",
      "|  20080101|   ABQ|           WN|     3425|      1440|         YES|       2|        120|\n",
      "|  20080101|   ABQ|           WN|     3481|      1500|         YES|      78|        210|\n",
      "|  20080101|   ABQ|           WN|     1135|      1500|         YES|      21|        210|\n",
      "|  20080101|   ABQ|           WN|      901|      1520|         YES|      18|        189|\n",
      "|  20080101|   ABQ|           WN|     2284|      1520|         YES|       3|        189|\n",
      "|  20080101|   ABQ|           WN|      357|      1525|         YES|     171|        360|\n",
      "|  20080101|   ABQ|           XE|     2922|      1530|         YES|      16|        383|\n",
      "|  20080101|   ABQ|           WN|      729|      1530|         YES|       7|        383|\n",
      "|  20080101|   ABQ|           WN|     1123|      1600|         YES|      15|        346|\n",
      "|  20080101|   ABQ|           AA|     1486|      1600|         YES|      17|        346|\n",
      "|  20080101|   ABQ|           OO|     6416|      1610|         YES|       8|        255|\n",
      "|  20080101|   ABQ|           WN|      397|      1630|         YES|       9|         72|\n",
      "|  20080101|   ABQ|           WN|      294|      1645|         YES|      11|         60|\n",
      "|  20080101|   ABQ|           XE|     2018|      1705|         YES|      10|         38|\n",
      "|  20080101|   ABQ|           WN|     2882|      1710|         YES|      16|         54|\n",
      "|  20080101|   ABQ|           WN|      644|      1725|         YES|       5|         51|\n",
      "|  20080101|   ABQ|           WN|     2788|      1755|         YES|      53|         84|\n",
      "|  20080101|   ABQ|           WN|      360|      1800|         YES|       5|         89|\n",
      "|  20080101|   ABQ|           WN|      972|      1810|         YES|       1|         80|\n",
      "|  20080101|   ABQ|           WN|     2641|      1815|         YES|      19|         83|\n",
      "|  20080101|   ABQ|           WN|     2497|      1825|         YES|      93|        176|\n",
      "|  20080101|   ABQ|           WN|     1114|      1845|         YES|      14|        279|\n",
      "|  20080101|   ABQ|           WN|     1178|      1845|         YES|      94|        279|\n",
      "|  20080101|   ABQ|           XE|      221|      1850|         YES|      68|        347|\n",
      "|  20080101|   ABQ|           WN|     2010|      1910|         YES|      15|        304|\n",
      "+----------+------+-------------+---------+----------+------------+--------+-----------+\n",
      "only showing top 50 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(\"DepDelaySum\", sum(\"DepDelay\").over(spec)). \\\n",
    "    orderBy(\"FlightDate\", \"Origin\", \"CRSDepTime\"). \\\n",
    "    show(50)"
   ]
  }
 ],
 "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
}