Pyspark- Windows_Ranks_Aggs.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "02eb6a99",
   "metadata": {},
   "source": [
    "#### This notebook is about advanced transformations and windowing in pyspark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "id": "fe76488c",
   "metadata": {},
   "outputs": [],
   "source": [
    "import pyspark\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark import SparkConf, SparkContext"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "b0b61326",
   "metadata": {},
   "outputs": [],
   "source": [
    "productPath = \"/home/solverbot/spark-warehouse/retail_db/products/part-00000\"\n",
    "orderitemPath = \"/home/solverbot/spark-warehouse/retail_db/order_items/part-00000\"\n",
    "ordersPath = \"/home/solverbot/spark-warehouse/retail_db/orders/part-00000.txt\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "id": "baa1b48d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "22/11/22 07:07:21 WARN Utils: Your hostname, codeStation resolves to a loopback address: 127.0.1.1; using 172.17.0.1 instead (on interface docker0)\n",
      "22/11/22 07:07:21 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/22 07:07:24 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable\n"
     ]
    }
   ],
   "source": [
    "#What is the difference between Session and Context?\n",
    "#SC is part of the Spark session that is established above\n",
    "spark = SparkSession.builder.appName('Adv Transformations').getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "id": "f609fb46",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://172.17.0.1:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.3.1</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>Adv Transformations</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<SparkContext master=local[*] appName=Adv Transformations>"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc = spark.sparkContext\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "id": "e5e70e52",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "id": "92622c04",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "orderItemDF = spark.read.csv(\"/home/solverbot/spark-warehouse/retail_db/order_items/\",inferSchema=True). \\\n",
    "            toDF(\"order_item_id\",\"order_item_order_id\",\"product_id\", \"qty\",\"product_cost\",\"order_subtotal\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "25050fca",
   "metadata": {},
   "outputs": [],
   "source": [
    "orderDF = spark.read.csv(\"/home/solverbot/spark-warehouse/retail_db/orders\",inferSchema=True) \\\n",
    "            .toDF(\"order_id\",\"order_date\",\"order_customer_id\",\"order_status\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "885564ce",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------------+-----------------+---------------+\n",
      "|order_id|         order_date|order_customer_id|   order_status|\n",
      "+--------+-------------------+-----------------+---------------+\n",
      "|       1|2013-07-25 00:00:00|            11599|         CLOSED|\n",
      "|       2|2013-07-25 00:00:00|              256|PENDING_PAYMENT|\n",
      "+--------+-------------------+-----------------+---------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orderDF.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "8f3db55d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-------------------+----------+---+------------+--------------+\n",
      "|order_item_id|order_item_order_id|product_id|qty|product_cost|order_subtotal|\n",
      "+-------------+-------------------+----------+---+------------+--------------+\n",
      "|            1|                  1|       957|  1|      299.98|        299.98|\n",
      "|            2|                  2|      1073|  1|      199.99|        199.99|\n",
      "+-------------+-------------------+----------+---+------------+--------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orderItemDF.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "73cb783c",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.window import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "id": "1689dfee",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on module pyspark.sql.window in pyspark.sql:\n",
      "\n",
      "NAME\n",
      "    pyspark.sql.window\n",
      "\n",
      "DESCRIPTION\n",
      "    # Licensed to the Apache Software Foundation (ASF) under one or more\n",
      "    # contributor license agreements.  See the NOTICE file distributed with\n",
      "    # this work for additional information regarding copyright ownership.\n",
      "    # The ASF licenses this file to You under the Apache License, Version 2.0\n",
      "    # (the \"License\"); you may not use this file except in compliance with\n",
      "    # the License.  You may obtain a copy of the License at\n",
      "    #\n",
      "    #    http://www.apache.org/licenses/LICENSE-2.0\n",
      "    #\n",
      "    # Unless required by applicable law or agreed to in writing, software\n",
      "    # distributed under the License is distributed on an \"AS IS\" BASIS,\n",
      "    # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
      "    # See the License for the specific language governing permissions and\n",
      "    # limitations under the License.\n",
      "    #\n",
      "\n",
      "CLASSES\n",
      "    builtins.object\n",
      "        Window\n",
      "        WindowSpec\n",
      "    \n",
      "    class Window(builtins.object)\n",
      "     |  Utility functions for defining window in DataFrames.\n",
      "     |  \n",
      "     |  .. versionadded:: 1.4\n",
      "     |  \n",
      "     |  Notes\n",
      "     |  -----\n",
      "     |  When ordering is not defined, an unbounded window frame (rowFrame,\n",
      "     |  unboundedPreceding, unboundedFollowing) is used by default. When ordering is defined,\n",
      "     |  a growing window frame (rangeFrame, unboundedPreceding, currentRow) is used by default.\n",
      "     |  \n",
      "     |  Examples\n",
      "     |  --------\n",
      "     |  >>> # ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW\n",
      "     |  >>> window = Window.orderBy(\"date\").rowsBetween(Window.unboundedPreceding, Window.currentRow)\n",
      "     |  \n",
      "     |  >>> # PARTITION BY country ORDER BY date RANGE BETWEEN 3 PRECEDING AND 3 FOLLOWING\n",
      "     |  >>> window = Window.orderBy(\"date\").partitionBy(\"country\").rangeBetween(-3, 3)\n",
      "     |  \n",
      "     |  Static methods defined here:\n",
      "     |  \n",
      "     |  orderBy(*cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      "     |      Creates a :class:`WindowSpec` with the ordering defined.\n",
      "     |      \n",
      "     |      .. versionadded:: 1.4\n",
      "     |  \n",
      "     |  partitionBy(*cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      "     |      Creates a :class:`WindowSpec` with the partitioning defined.\n",
      "     |      \n",
      "     |      .. versionadded:: 1.4\n",
      "     |  \n",
      "     |  rangeBetween(start: int, end: int) -> 'WindowSpec'\n",
      "     |      Creates a :class:`WindowSpec` with the frame boundaries defined,\n",
      "     |      from `start` (inclusive) to `end` (inclusive).\n",
      "     |      \n",
      "     |      Both `start` and `end` are relative from the current row. For example,\n",
      "     |      \"0\" means \"current row\", while \"-1\" means one off before the current row,\n",
      "     |      and \"5\" means the five off after the current row.\n",
      "     |      \n",
      "     |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      "     |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      "     |      values directly.\n",
      "     |      \n",
      "     |      A range-based boundary is based on the actual value of the ORDER BY\n",
      "     |      expression(s). An offset is used to alter the value of the ORDER BY expression, for\n",
      "     |      instance if the current ORDER BY expression has a value of 10 and the lower bound offset\n",
      "     |      is -3, the resulting lower bound for the current row will be 10 - 3 = 7. This however puts a\n",
      "     |      number of constraints on the ORDER BY expressions: there can be only one expression and this\n",
      "     |      expression must have a numerical data type. An exception can be made when the offset is\n",
      "     |      unbounded, because no value modification is needed, in this case multiple and non-numeric\n",
      "     |      ORDER BY expression are allowed.\n",
      "     |      \n",
      "     |      .. versionadded:: 2.1.0\n",
      "     |      \n",
      "     |      Parameters\n",
      "     |      ----------\n",
      "     |      start : int\n",
      "     |          boundary start, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      "     |          any value less than or equal to max(-sys.maxsize, -9223372036854775808).\n",
      "     |      end : int\n",
      "     |          boundary end, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      "     |          any value greater than or equal to min(sys.maxsize, 9223372036854775807).\n",
      "     |      \n",
      "     |      Examples\n",
      "     |      --------\n",
      "     |      >>> from pyspark.sql import Window\n",
      "     |      >>> from pyspark.sql import functions as func\n",
      "     |      >>> from pyspark.sql import SQLContext\n",
      "     |      >>> sc = SparkContext.getOrCreate()\n",
      "     |      >>> sqlContext = SQLContext(sc)\n",
      "     |      >>> tup = [(1, \"a\"), (1, \"a\"), (2, \"a\"), (1, \"b\"), (2, \"b\"), (3, \"b\")]\n",
      "     |      >>> df = sqlContext.createDataFrame(tup, [\"id\", \"category\"])\n",
      "     |      >>> window = Window.partitionBy(\"category\").orderBy(\"id\").rangeBetween(Window.currentRow, 1)\n",
      "     |      >>> df.withColumn(\"sum\", func.sum(\"id\").over(window)).sort(\"id\", \"category\").show()\n",
      "     |      +---+--------+---+\n",
      "     |      | id|category|sum|\n",
      "     |      +---+--------+---+\n",
      "     |      |  1|       a|  4|\n",
      "     |      |  1|       a|  4|\n",
      "     |      |  1|       b|  3|\n",
      "     |      |  2|       a|  2|\n",
      "     |      |  2|       b|  5|\n",
      "     |      |  3|       b|  3|\n",
      "     |      +---+--------+---+\n",
      "     |  \n",
      "     |  rowsBetween(start: int, end: int) -> 'WindowSpec'\n",
      "     |      Creates a :class:`WindowSpec` with the frame boundaries defined,\n",
      "     |      from `start` (inclusive) to `end` (inclusive).\n",
      "     |      \n",
      "     |      Both `start` and `end` are relative positions from the current row.\n",
      "     |      For example, \"0\" means \"current row\", while \"-1\" means the row before\n",
      "     |      the current row, and \"5\" means the fifth row after the current row.\n",
      "     |      \n",
      "     |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      "     |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      "     |      values directly.\n",
      "     |      \n",
      "     |      A row based boundary is based on the position of the row within the partition.\n",
      "     |      An offset indicates the number of rows above or below the current row, the frame for the\n",
      "     |      current row starts or ends. For instance, given a row based sliding frame with a lower bound\n",
      "     |      offset of -1 and a upper bound offset of +2. The frame for row with index 5 would range from\n",
      "     |      index 4 to index 7.\n",
      "     |      \n",
      "     |      .. versionadded:: 2.1.0\n",
      "     |      \n",
      "     |      Parameters\n",
      "     |      ----------\n",
      "     |      start : int\n",
      "     |          boundary start, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      "     |          any value less than or equal to -9223372036854775808.\n",
      "     |      end : int\n",
      "     |          boundary end, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      "     |          any value greater than or equal to 9223372036854775807.\n",
      "     |      \n",
      "     |      Examples\n",
      "     |      --------\n",
      "     |      >>> from pyspark.sql import Window\n",
      "     |      >>> from pyspark.sql import functions as func\n",
      "     |      >>> from pyspark.sql import SQLContext\n",
      "     |      >>> sc = SparkContext.getOrCreate()\n",
      "     |      >>> sqlContext = SQLContext(sc)\n",
      "     |      >>> tup = [(1, \"a\"), (1, \"a\"), (2, \"a\"), (1, \"b\"), (2, \"b\"), (3, \"b\")]\n",
      "     |      >>> df = sqlContext.createDataFrame(tup, [\"id\", \"category\"])\n",
      "     |      >>> window = Window.partitionBy(\"category\").orderBy(\"id\").rowsBetween(Window.currentRow, 1)\n",
      "     |      >>> df.withColumn(\"sum\", func.sum(\"id\").over(window)).sort(\"id\", \"category\", \"sum\").show()\n",
      "     |      +---+--------+---+\n",
      "     |      | id|category|sum|\n",
      "     |      +---+--------+---+\n",
      "     |      |  1|       a|  2|\n",
      "     |      |  1|       a|  3|\n",
      "     |      |  1|       b|  3|\n",
      "     |      |  2|       a|  2|\n",
      "     |      |  2|       b|  5|\n",
      "     |      |  3|       b|  3|\n",
      "     |      +---+--------+---+\n",
      "     |  \n",
      "     |  ----------------------------------------------------------------------\n",
      "     |  Data descriptors defined here:\n",
      "     |  \n",
      "     |  __dict__\n",
      "     |      dictionary for instance variables (if defined)\n",
      "     |  \n",
      "     |  __weakref__\n",
      "     |      list of weak references to the object (if defined)\n",
      "     |  \n",
      "     |  ----------------------------------------------------------------------\n",
      "     |  Data and other attributes defined here:\n",
      "     |  \n",
      "     |  __annotations__ = {'currentRow': <class 'int'>, 'unboundedFollowing': ...\n",
      "     |  \n",
      "     |  currentRow = 0\n",
      "     |  \n",
      "     |  unboundedFollowing = 9223372036854775807\n",
      "     |  \n",
      "     |  unboundedPreceding = -9223372036854775808\n",
      "    \n",
      "    class WindowSpec(builtins.object)\n",
      "     |  WindowSpec(jspec: py4j.java_gateway.JavaObject) -> None\n",
      "     |  \n",
      "     |  A window specification that defines the partitioning, ordering,\n",
      "     |  and frame boundaries.\n",
      "     |  \n",
      "     |  Use the static methods in :class:`Window` to create a :class:`WindowSpec`.\n",
      "     |  \n",
      "     |  .. versionadded:: 1.4.0\n",
      "     |  \n",
      "     |  Methods defined here:\n",
      "     |  \n",
      "     |  __init__(self, jspec: py4j.java_gateway.JavaObject) -> None\n",
      "     |      Initialize self.  See help(type(self)) for accurate signature.\n",
      "     |  \n",
      "     |  orderBy(self, *cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      "     |      Defines the ordering columns in a :class:`WindowSpec`.\n",
      "     |      \n",
      "     |      .. versionadded:: 1.4.0\n",
      "     |      \n",
      "     |      Parameters\n",
      "     |      ----------\n",
      "     |      cols : str, :class:`Column` or list\n",
      "     |          names of columns or expressions\n",
      "     |  \n",
      "     |  partitionBy(self, *cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      "     |      Defines the partitioning columns in a :class:`WindowSpec`.\n",
      "     |      \n",
      "     |      .. versionadded:: 1.4.0\n",
      "     |      \n",
      "     |      Parameters\n",
      "     |      ----------\n",
      "     |      cols : str, :class:`Column` or list\n",
      "     |          names of columns or expressions\n",
      "     |  \n",
      "     |  rangeBetween(self, start: int, end: int) -> 'WindowSpec'\n",
      "     |      Defines the frame boundaries, from `start` (inclusive) to `end` (inclusive).\n",
      "     |      \n",
      "     |      Both `start` and `end` are relative from the current row. For example,\n",
      "     |      \"0\" means \"current row\", while \"-1\" means one off before the current row,\n",
      "     |      and \"5\" means the five off after the current row.\n",
      "     |      \n",
      "     |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      "     |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      "     |      values directly.\n",
      "     |      \n",
      "     |      .. versionadded:: 1.4.0\n",
      "     |      \n",
      "     |      Parameters\n",
      "     |      ----------\n",
      "     |      start : int\n",
      "     |          boundary start, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      "     |          any value less than or equal to max(-sys.maxsize, -9223372036854775808).\n",
      "     |      end : int\n",
      "     |          boundary end, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      "     |          any value greater than or equal to min(sys.maxsize, 9223372036854775807).\n",
      "     |  \n",
      "     |  rowsBetween(self, start: int, end: int) -> 'WindowSpec'\n",
      "     |      Defines the frame boundaries, from `start` (inclusive) to `end` (inclusive).\n",
      "     |      \n",
      "     |      Both `start` and `end` are relative positions from the current row.\n",
      "     |      For example, \"0\" means \"current row\", while \"-1\" means the row before\n",
      "     |      the current row, and \"5\" means the fifth row after the current row.\n",
      "     |      \n",
      "     |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      "     |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      "     |      values directly.\n",
      "     |      \n",
      "     |      .. versionadded:: 1.4.0\n",
      "     |      \n",
      "     |      Parameters\n",
      "     |      ----------\n",
      "     |      start : int\n",
      "     |          boundary start, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      "     |          any value less than or equal to max(-sys.maxsize, -9223372036854775808).\n",
      "     |      end : int\n",
      "     |          boundary end, inclusive.\n",
      "     |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      "     |          any value greater than or equal to min(sys.maxsize, 9223372036854775807).\n",
      "     |  \n",
      "     |  ----------------------------------------------------------------------\n",
      "     |  Data descriptors defined here:\n",
      "     |  \n",
      "     |  __dict__\n",
      "     |      dictionary for instance variables (if defined)\n",
      "     |  \n",
      "     |  __weakref__\n",
      "     |      list of weak references to the object (if defined)\n",
      "\n",
      "DATA\n",
      "    __all__ = ['Window', 'WindowSpec']\n",
      "\n",
      "FILE\n",
      "    /home/solverbot/.local/lib/python3.10/site-packages/pyspark/sql/window.py\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(pyspark.sql.window)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "f895b1cf",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on class Window in module pyspark.sql.window:\n",
      "\n",
      "class Window(builtins.object)\n",
      " |  Utility functions for defining window in DataFrames.\n",
      " |  \n",
      " |  .. versionadded:: 1.4\n",
      " |  \n",
      " |  Notes\n",
      " |  -----\n",
      " |  When ordering is not defined, an unbounded window frame (rowFrame,\n",
      " |  unboundedPreceding, unboundedFollowing) is used by default. When ordering is defined,\n",
      " |  a growing window frame (rangeFrame, unboundedPreceding, currentRow) is used by default.\n",
      " |  \n",
      " |  Examples\n",
      " |  --------\n",
      " |  >>> # ORDER BY date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW\n",
      " |  >>> window = Window.orderBy(\"date\").rowsBetween(Window.unboundedPreceding, Window.currentRow)\n",
      " |  \n",
      " |  >>> # PARTITION BY country ORDER BY date RANGE BETWEEN 3 PRECEDING AND 3 FOLLOWING\n",
      " |  >>> window = Window.orderBy(\"date\").partitionBy(\"country\").rangeBetween(-3, 3)\n",
      " |  \n",
      " |  Static methods defined here:\n",
      " |  \n",
      " |  orderBy(*cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      " |      Creates a :class:`WindowSpec` with the ordering defined.\n",
      " |      \n",
      " |      .. versionadded:: 1.4\n",
      " |  \n",
      " |  partitionBy(*cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      " |      Creates a :class:`WindowSpec` with the partitioning defined.\n",
      " |      \n",
      " |      .. versionadded:: 1.4\n",
      " |  \n",
      " |  rangeBetween(start: int, end: int) -> 'WindowSpec'\n",
      " |      Creates a :class:`WindowSpec` with the frame boundaries defined,\n",
      " |      from `start` (inclusive) to `end` (inclusive).\n",
      " |      \n",
      " |      Both `start` and `end` are relative from the current row. For example,\n",
      " |      \"0\" means \"current row\", while \"-1\" means one off before the current row,\n",
      " |      and \"5\" means the five off after the current row.\n",
      " |      \n",
      " |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      " |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      " |      values directly.\n",
      " |      \n",
      " |      A range-based boundary is based on the actual value of the ORDER BY\n",
      " |      expression(s). An offset is used to alter the value of the ORDER BY expression, for\n",
      " |      instance if the current ORDER BY expression has a value of 10 and the lower bound offset\n",
      " |      is -3, the resulting lower bound for the current row will be 10 - 3 = 7. This however puts a\n",
      " |      number of constraints on the ORDER BY expressions: there can be only one expression and this\n",
      " |      expression must have a numerical data type. An exception can be made when the offset is\n",
      " |      unbounded, because no value modification is needed, in this case multiple and non-numeric\n",
      " |      ORDER BY expression are allowed.\n",
      " |      \n",
      " |      .. versionadded:: 2.1.0\n",
      " |      \n",
      " |      Parameters\n",
      " |      ----------\n",
      " |      start : int\n",
      " |          boundary start, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      " |          any value less than or equal to max(-sys.maxsize, -9223372036854775808).\n",
      " |      end : int\n",
      " |          boundary end, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      " |          any value greater than or equal to min(sys.maxsize, 9223372036854775807).\n",
      " |      \n",
      " |      Examples\n",
      " |      --------\n",
      " |      >>> from pyspark.sql import Window\n",
      " |      >>> from pyspark.sql import functions as func\n",
      " |      >>> from pyspark.sql import SQLContext\n",
      " |      >>> sc = SparkContext.getOrCreate()\n",
      " |      >>> sqlContext = SQLContext(sc)\n",
      " |      >>> tup = [(1, \"a\"), (1, \"a\"), (2, \"a\"), (1, \"b\"), (2, \"b\"), (3, \"b\")]\n",
      " |      >>> df = sqlContext.createDataFrame(tup, [\"id\", \"category\"])\n",
      " |      >>> window = Window.partitionBy(\"category\").orderBy(\"id\").rangeBetween(Window.currentRow, 1)\n",
      " |      >>> df.withColumn(\"sum\", func.sum(\"id\").over(window)).sort(\"id\", \"category\").show()\n",
      " |      +---+--------+---+\n",
      " |      | id|category|sum|\n",
      " |      +---+--------+---+\n",
      " |      |  1|       a|  4|\n",
      " |      |  1|       a|  4|\n",
      " |      |  1|       b|  3|\n",
      " |      |  2|       a|  2|\n",
      " |      |  2|       b|  5|\n",
      " |      |  3|       b|  3|\n",
      " |      +---+--------+---+\n",
      " |  \n",
      " |  rowsBetween(start: int, end: int) -> 'WindowSpec'\n",
      " |      Creates a :class:`WindowSpec` with the frame boundaries defined,\n",
      " |      from `start` (inclusive) to `end` (inclusive).\n",
      " |      \n",
      " |      Both `start` and `end` are relative positions from the current row.\n",
      " |      For example, \"0\" means \"current row\", while \"-1\" means the row before\n",
      " |      the current row, and \"5\" means the fifth row after the current row.\n",
      " |      \n",
      " |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      " |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      " |      values directly.\n",
      " |      \n",
      " |      A row based boundary is based on the position of the row within the partition.\n",
      " |      An offset indicates the number of rows above or below the current row, the frame for the\n",
      " |      current row starts or ends. For instance, given a row based sliding frame with a lower bound\n",
      " |      offset of -1 and a upper bound offset of +2. The frame for row with index 5 would range from\n",
      " |      index 4 to index 7.\n",
      " |      \n",
      " |      .. versionadded:: 2.1.0\n",
      " |      \n",
      " |      Parameters\n",
      " |      ----------\n",
      " |      start : int\n",
      " |          boundary start, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      " |          any value less than or equal to -9223372036854775808.\n",
      " |      end : int\n",
      " |          boundary end, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      " |          any value greater than or equal to 9223372036854775807.\n",
      " |      \n",
      " |      Examples\n",
      " |      --------\n",
      " |      >>> from pyspark.sql import Window\n",
      " |      >>> from pyspark.sql import functions as func\n",
      " |      >>> from pyspark.sql import SQLContext\n",
      " |      >>> sc = SparkContext.getOrCreate()\n",
      " |      >>> sqlContext = SQLContext(sc)\n",
      " |      >>> tup = [(1, \"a\"), (1, \"a\"), (2, \"a\"), (1, \"b\"), (2, \"b\"), (3, \"b\")]\n",
      " |      >>> df = sqlContext.createDataFrame(tup, [\"id\", \"category\"])\n",
      " |      >>> window = Window.partitionBy(\"category\").orderBy(\"id\").rowsBetween(Window.currentRow, 1)\n",
      " |      >>> df.withColumn(\"sum\", func.sum(\"id\").over(window)).sort(\"id\", \"category\", \"sum\").show()\n",
      " |      +---+--------+---+\n",
      " |      | id|category|sum|\n",
      " |      +---+--------+---+\n",
      " |      |  1|       a|  2|\n",
      " |      |  1|       a|  3|\n",
      " |      |  1|       b|  3|\n",
      " |      |  2|       a|  2|\n",
      " |      |  2|       b|  5|\n",
      " |      |  3|       b|  3|\n",
      " |      +---+--------+---+\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Data descriptors defined here:\n",
      " |  \n",
      " |  __dict__\n",
      " |      dictionary for instance variables (if defined)\n",
      " |  \n",
      " |  __weakref__\n",
      " |      list of weak references to the object (if defined)\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Data and other attributes defined here:\n",
      " |  \n",
      " |  __annotations__ = {'currentRow': <class 'int'>, 'unboundedFollowing': ...\n",
      " |  \n",
      " |  currentRow = 0\n",
      " |  \n",
      " |  unboundedFollowing = 9223372036854775807\n",
      " |  \n",
      " |  unboundedPreceding = -9223372036854775808\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(Window)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "id": "5dbccf24",
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Help on class WindowSpec in module pyspark.sql.window:\n",
      "\n",
      "class WindowSpec(builtins.object)\n",
      " |  WindowSpec(jspec: py4j.java_gateway.JavaObject) -> None\n",
      " |  \n",
      " |  A window specification that defines the partitioning, ordering,\n",
      " |  and frame boundaries.\n",
      " |  \n",
      " |  Use the static methods in :class:`Window` to create a :class:`WindowSpec`.\n",
      " |  \n",
      " |  .. versionadded:: 1.4.0\n",
      " |  \n",
      " |  Methods defined here:\n",
      " |  \n",
      " |  __init__(self, jspec: py4j.java_gateway.JavaObject) -> None\n",
      " |      Initialize self.  See help(type(self)) for accurate signature.\n",
      " |  \n",
      " |  orderBy(self, *cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      " |      Defines the ordering columns in a :class:`WindowSpec`.\n",
      " |      \n",
      " |      .. versionadded:: 1.4.0\n",
      " |      \n",
      " |      Parameters\n",
      " |      ----------\n",
      " |      cols : str, :class:`Column` or list\n",
      " |          names of columns or expressions\n",
      " |  \n",
      " |  partitionBy(self, *cols: Union[ForwardRef('ColumnOrName'), List[ForwardRef('ColumnOrName_')]]) -> 'WindowSpec'\n",
      " |      Defines the partitioning columns in a :class:`WindowSpec`.\n",
      " |      \n",
      " |      .. versionadded:: 1.4.0\n",
      " |      \n",
      " |      Parameters\n",
      " |      ----------\n",
      " |      cols : str, :class:`Column` or list\n",
      " |          names of columns or expressions\n",
      " |  \n",
      " |  rangeBetween(self, start: int, end: int) -> 'WindowSpec'\n",
      " |      Defines the frame boundaries, from `start` (inclusive) to `end` (inclusive).\n",
      " |      \n",
      " |      Both `start` and `end` are relative from the current row. For example,\n",
      " |      \"0\" means \"current row\", while \"-1\" means one off before the current row,\n",
      " |      and \"5\" means the five off after the current row.\n",
      " |      \n",
      " |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      " |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      " |      values directly.\n",
      " |      \n",
      " |      .. versionadded:: 1.4.0\n",
      " |      \n",
      " |      Parameters\n",
      " |      ----------\n",
      " |      start : int\n",
      " |          boundary start, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      " |          any value less than or equal to max(-sys.maxsize, -9223372036854775808).\n",
      " |      end : int\n",
      " |          boundary end, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      " |          any value greater than or equal to min(sys.maxsize, 9223372036854775807).\n",
      " |  \n",
      " |  rowsBetween(self, start: int, end: int) -> 'WindowSpec'\n",
      " |      Defines the frame boundaries, from `start` (inclusive) to `end` (inclusive).\n",
      " |      \n",
      " |      Both `start` and `end` are relative positions from the current row.\n",
      " |      For example, \"0\" means \"current row\", while \"-1\" means the row before\n",
      " |      the current row, and \"5\" means the fifth row after the current row.\n",
      " |      \n",
      " |      We recommend users use ``Window.unboundedPreceding``, ``Window.unboundedFollowing``,\n",
      " |      and ``Window.currentRow`` to specify special boundary values, rather than using integral\n",
      " |      values directly.\n",
      " |      \n",
      " |      .. versionadded:: 1.4.0\n",
      " |      \n",
      " |      Parameters\n",
      " |      ----------\n",
      " |      start : int\n",
      " |          boundary start, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedPreceding``, or\n",
      " |          any value less than or equal to max(-sys.maxsize, -9223372036854775808).\n",
      " |      end : int\n",
      " |          boundary end, inclusive.\n",
      " |          The frame is unbounded if this is ``Window.unboundedFollowing``, or\n",
      " |          any value greater than or equal to min(sys.maxsize, 9223372036854775807).\n",
      " |  \n",
      " |  ----------------------------------------------------------------------\n",
      " |  Data descriptors defined here:\n",
      " |  \n",
      " |  __dict__\n",
      " |      dictionary for instance variables (if defined)\n",
      " |  \n",
      " |  __weakref__\n",
      " |      list of weak references to the object (if defined)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "help(WindowSpec)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "id": "479bd4c2",
   "metadata": {},
   "outputs": [],
   "source": [
    "o = orderDF.alias(\"o\")\n",
    "oi = orderItemDF.alias('oi')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "id": "2eb06c9d",
   "metadata": {},
   "outputs": [],
   "source": [
    "#Creating temp view\n",
    "oi.groupBy(\"product_id\").agg(sum('order_subtotal').alias('product_revn')).createTempView('product_revenue')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "id": "2d96fc9d",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+\n",
      "|product_id|     product_revn|\n",
      "+----------+-----------------+\n",
      "|       897|7022.189999999962|\n",
      "|       804| 6136.92999999996|\n",
      "+----------+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "spark.sql(\"SELECT * FROM product_revenue\").show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "id": "dbb8cd98",
   "metadata": {},
   "outputs": [],
   "source": [
    "order_revenue = oi.groupBy(\"product_id\").agg(sum('order_subtotal').alias('product_revn')).alias(\"order_revenue\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "id": "853b578a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+\n",
      "|product_id|     product_revn|\n",
      "+----------+-----------------+\n",
      "|       897|7022.189999999962|\n",
      "|       804| 6136.92999999996|\n",
      "+----------+-----------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "order_revenue.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "id": "5eb772f0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----+--------+--------------------+--------+--------------------+\n",
      "|p_id|categ_id|            pdt_name|pdt_cost|            pdt_link|\n",
      "+----+--------+--------------------+--------+--------------------+\n",
      "|   1|       2|Quest Q64 10 FT. ...|   59.98|http://images.acm...|\n",
      "|   2|       2|Under Armour Men'...|  129.99|http://images.acm...|\n",
      "+----+--------+--------------------+--------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "products = spark.read.csv(\"/home/solverbot/spark-warehouse/retail_db/products/\",inferSchema=True) \\\n",
    "                .toDF(\"p_id\",\"categ_id\",\"pdt_name\",\"c3\",\"pdt_cost\",\"pdt_link\")\n",
    "products = products.drop(\"c3\")\n",
    "products.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "id": "66e440f3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "|product_id|     product_revn|categ_id|            pdt_name|pdt_cost|            pdt_link|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "|       897|7022.189999999962|      40|Team Golf New Eng...|   24.99|http://images.acm...|\n",
      "|       804| 6136.92999999996|      36|Glove It Women's ...|   19.99|http://images.acm...|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinPdtOrder = order_revenue.join(products, products.p_id == order_revenue.product_id).drop('p_id')\n",
    "joinPdtOrder.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "id": "bda058da",
   "metadata": {},
   "outputs": [],
   "source": [
    "categoryTables = joinPdtOrder.groupBy('categ_id').agg(count(col(\"product_id\")).alias('CategCount')) \\\n",
    "    .orderBy(col('CategCount').desc())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "id": "9af3b1a1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "33"
      ]
     },
     "execution_count": 61,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "categoryTables.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "id": "4cbfc570",
   "metadata": {},
   "outputs": [],
   "source": [
    "spec = Window.partitionBy(joinPdtOrder.categ_id)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "id": "0e0cc709",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "pyspark.sql.window.WindowSpec"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "type(spec)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "id": "f0250f2c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "|product_id|     product_revn|categ_id|            pdt_name|pdt_cost|            pdt_link|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "|        19| 7999.35999999999|       2|Nike Men's Finger...|  124.99|http://images.acm...|\n",
      "|        24|5919.259999999989|       2|Elevation Trainin...|   79.99|http://images.acm...|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 112:============================>                            (1 + 1) / 2]\r",
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "joinPdtOrder.where('categ_id = 2').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "id": "34d4a95a",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+------------------+\n",
      "|categ_id|      categ_window|\n",
      "+--------+------------------+\n",
      "|      16|  6799.31999999999|\n",
      "|      34| 10369.38999999999|\n",
      "|       5|13296.569999999952|\n",
      "|       2|13918.619999999979|\n",
      "|       6|14756.719999999943|\n",
      "|      11|16461.909999999953|\n",
      "|       7|22660.729999999996|\n",
      "|      32| 23295.58999999994|\n",
      "|      31|24188.189999999973|\n",
      "|      44| 24329.16999999994|\n",
      "|      36| 25924.24999999983|\n",
      "|       4| 27099.32999999999|\n",
      "|      12| 28365.02999999991|\n",
      "|      30|33337.169999999955|\n",
      "|      41| 35362.25999999991|\n",
      "|       3| 37863.67999999992|\n",
      "|      13| 38688.72999999983|\n",
      "|      33|           40873.0|\n",
      "|      40| 44482.19999999975|\n",
      "|      26|50141.999999999956|\n",
      "+--------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinPdtOrder.select(\"categ_id\", sum(joinPdtOrder.product_revn). \\\n",
    "                    over(spec). \\\n",
    "                    alias('categ_window')). \\\n",
    "                    distinct(). \\\n",
    "                    orderBy('categ_window'). \\\n",
    "                    show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "id": "a3b4e064",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------------------+--------+--------------------+--------+--------------------+---------+---------+\n",
      "|product_id|      product_revn|categ_id|            pdt_name|pdt_cost|            pdt_link|categ_sum|categ_avg|\n",
      "+----------+------------------+--------+--------------------+--------+--------------------+---------+---------+\n",
      "|        19|  7999.35999999999|       2|Nike Men's Finger...|  124.99|http://images.acm...|  13919.0|   6959.0|\n",
      "|        24| 5919.259999999989|       2|Elevation Trainin...|   79.99|http://images.acm...|  13919.0|   6959.0|\n",
      "|        44|18296.949999999964|       3|adidas Men's F10 ...|   59.99|http://images.acm...|  37864.0|  12621.0|\n",
      "+----------+------------------+--------+--------------------+--------+--------------------+---------+---------+\n",
      "only showing top 3 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinPdtOrder.withColumn(\"categ_sum\", round(sum(joinPdtOrder.product_revn).over(spec))) \\\n",
    "            .withColumn(\"categ_avg\", round(avg(joinPdtOrder.product_revn).over(spec))) \\\n",
    "            .show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "id": "0f4459c3",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-------------------+----------+---+------------+--------------+\n",
      "|order_item_id|order_item_order_id|product_id|qty|product_cost|order_subtotal|\n",
      "+-------------+-------------------+----------+---+------------+--------------+\n",
      "|            1|                  1|       957|  1|      299.98|        299.98|\n",
      "|            2|                  2|      1073|  1|      199.99|        199.99|\n",
      "|            3|                  2|       502|  5|       250.0|          50.0|\n",
      "|            4|                  2|       403|  1|      129.99|        129.99|\n",
      "|            5|                  4|       897|  2|       49.98|         24.99|\n",
      "+-------------+-------------------+----------+---+------------+--------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "orderItemDF.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "id": "40f961db",
   "metadata": {},
   "outputs": [],
   "source": [
    "order_spec = Window.partitionBy('order_item_order_id'). \\\n",
    "                    orderBy(orderItemDF.order_subtotal.desc())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "id": "140c0120",
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "[Stage 166:============================>                            (1 + 1) / 2]\r"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------------+-------------------+----------+---+------------+--------------+------------+\n",
      "|order_item_id|order_item_order_id|product_id|qty|product_cost|order_subtotal|next_revenue|\n",
      "+-------------+-------------------+----------+---+------------+--------------+------------+\n",
      "|            1|                  1|       957|  1|      299.98|        299.98|        null|\n",
      "|            9|                  5|       957|  1|      299.98|        299.98|      299.98|\n",
      "|           12|                  5|       957|  1|      299.98|        299.98|      129.99|\n",
      "|           13|                  5|       403|  1|      129.99|        129.99|       59.99|\n",
      "|           10|                  5|       365|  5|      299.95|         59.99|       49.98|\n",
      "|           11|                  5|      1014|  2|       99.96|         49.98|        null|\n",
      "|           34|                 12|       957|  1|      299.98|        299.98|       99.99|\n",
      "|           37|                 12|       191|  5|      499.95|         99.99|        50.0|\n",
      "|           38|                 12|       502|  5|       250.0|          50.0|       49.98|\n",
      "|           36|                 12|      1014|  3|      149.94|         49.98|        25.0|\n",
      "|           35|                 12|       134|  4|       100.0|          25.0|        null|\n",
      "|           39|                 13|       276|  4|      127.96|         31.99|        null|\n",
      "|           47|                 15|      1004|  1|      399.98|        399.98|      199.99|\n",
      "|           44|                 15|      1073|  1|      199.99|        199.99|       59.99|\n",
      "|           46|                 15|       365|  3|      179.97|         59.99|        50.0|\n",
      "|           43|                 15|       502|  1|        50.0|          50.0|       31.99|\n",
      "|           45|                 15|       828|  3|       95.97|         31.99|        null|\n",
      "|           48|                 16|       365|  2|      119.98|         59.99|       59.99|\n",
      "|           49|                 16|       365|  5|      299.95|         59.99|        null|\n",
      "|           58|                 19|      1004|  1|      399.98|        399.98|      299.98|\n",
      "+-------------+-------------------+----------+---+------------+--------------+------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "\r",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "orderItemDF.withColumn('next_revenue',\n",
    "                    lead('order_subtotal').over(order_spec)) \\\n",
    "                    .show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "id": "d8e6bbd6",
   "metadata": {},
   "outputs": [],
   "source": [
    "rank_spec = Window.partitionBy(joinPdtOrder.categ_id).orderBy(joinPdtOrder.product_revn)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 84,
   "id": "879d1eda",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "|product_id|     product_revn|categ_id|            pdt_name|pdt_cost|            pdt_link|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "|       897|7022.189999999962|      40|Team Golf New Eng...|   24.99|http://images.acm...|\n",
      "|       804| 6136.92999999996|      36|Glove It Women's ...|   19.99|http://images.acm...|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "joinPdtOrder.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 87,
   "id": "9c85ec95",
   "metadata": {},
   "outputs": [],
   "source": [
    "pdtOrderRevnueRanked = joinPdtOrder.withColumn(\"rnk\", rank().over(rank_spec))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "id": "47de7001",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+--------+--------------------+--------+--------------------+---+\n",
      "|product_id|     product_revn|categ_id|            pdt_name|pdt_cost|            pdt_link|rnk|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+---+\n",
      "|        24|5919.259999999989|       2|Elevation Trainin...|   79.99|http://images.acm...|  1|\n",
      "|        19| 7999.35999999999|       2|Nike Men's Finger...|  124.99|http://images.acm...|  2|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+---+\n",
      "only showing top 2 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pdtOrderRevnueRanked.show(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "id": "fc6f3cb1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-----------------+--------+--------------------+--------+--------------------+---+\n",
      "|product_id|     product_revn|categ_id|            pdt_name|pdt_cost|            pdt_link|rnk|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+---+\n",
      "|        24|5919.259999999989|       2|Elevation Trainin...|   79.99|http://images.acm...|  1|\n",
      "|        19| 7999.35999999999|       2|Nike Men's Finger...|  124.99|http://images.acm...|  2|\n",
      "|        37|9167.379999999963|       3|adidas Kids' F5 M...|   34.99|http://images.acm...|  1|\n",
      "|        35|10399.34999999999|       3|adidas Brazuca 20...|  159.99|http://images.acm...|  2|\n",
      "|        61|8399.719999999996|       4|Diamondback Girls...|  299.99|http://images.acm...|  1|\n",
      "+----------+-----------------+--------+--------------------+--------+--------------------+---+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "pdtOrderRevnueRanked.where('rnk <= 2').show(5)"
   ]
  }
 ],
 "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
}