pyspark/3_rolling_window_features.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    " # Rolling Window Features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Following notebook showcases an example workflow of creating rolling window features and building a model to predict which customers will buy in next 4 weeks.\n",
    "\n",
    "This uses dummy sales data but the idea can be implemented on actual sales data and can also be expanded to include other available data sources such as click-stream data, call center data, email contacts data, etc.\n",
    "\n",
    "***\n",
    "\n",
    "<b>Spark 3.1.2</b> (with Python 3.8) has been used for this notebook.<br>\n",
    "Refer to [spark documentation](https://spark.apache.org/docs/3.1.2/api/sql/index.html) for help with <b>data ops functions</b>.<br>\n",
    "Refer to [this article](https://medium.com/analytics-vidhya/installing-and-using-pyspark-on-windows-machine-59c2d64af76e) to <b>install and use PySpark on Windows machine</b>."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Building a spark session\n",
    "To create a SparkSession, use the following builder pattern:\n",
    " \n",
    "`spark = SparkSession\\\n",
    "    .builder\\\n",
    "    .master(\"local\")\\\n",
    "    .appName(\"Word Count\")\\\n",
    "    .config(\"spark.some.config.option\", \"some-value\")\\\n",
    "    .getOrCreate()`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:25:21.626967Z",
     "start_time": "2021-12-28T06:25:21.318113Z"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql import functions as F\n",
    "from pyspark.sql import Window\n",
    "from pyspark.sql.types import FloatType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:25:28.141614Z",
     "start_time": "2021-12-28T06:25:27.634779Z"
    }
   },
   "outputs": [],
   "source": [
    "#initiating spark session\n",
    "spark.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:25:32.603308Z",
     "start_time": "2021-12-28T06:25:31.992692Z"
    }
   },
   "outputs": [],
   "source": [
    "spark = SparkSession\\\n",
    "    .builder\\\n",
    "    .appName(\"rolling_window\")\\\n",
    "    .config(\"spark.executor.memory\", \"1536m\")\\\n",
    "    .config(\"spark.driver.memory\", \"2g\")\\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:25:34.292160Z",
     "start_time": "2021-12-28T06:25:34.252288Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - hive</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://lenovo-pc:4041\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v2.4.5</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local[*]</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>rolling_window</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x1d95d3839c8>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Data prep\n",
    "\n",
    "We will be using window functions to compute relative features for all dates. We will first aggregate the data to customer x week level so it is easier to handle.\n",
    "\n",
    "<mark>The week level date that we create will serve as the 'reference date' from which everything will be relative.</mark>\n",
    "\n",
    "All the required dimension tables have to be joined with the sales table prior to aggregation so that we can create all required features."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Read input datasets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:26:58.796270Z",
     "start_time": "2021-12-28T06:26:56.691183Z"
    }
   },
   "outputs": [],
   "source": [
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:27:43.428892Z",
     "start_time": "2021-12-28T06:27:00.103941Z"
    }
   },
   "outputs": [],
   "source": [
    "df_sales = spark.read.csv('./data/rw_sales.csv',inferSchema=True,header=True)\n",
    "df_customer = spark.read.csv('./data/clustering_customer.csv',inferSchema=True,header=True)\n",
    "df_product = spark.read.csv('./data/clustering_product.csv',inferSchema=True,header=True)\n",
    "df_payment = spark.read.csv('./data/clustering_payment.csv',inferSchema=True,header=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Quick exploration of the datasets:</b>\n",
    "1. We have sales data that captures date, customer id, product, quantity, dollar amount & payment type at order x item level. `order_item_id` refers to each unique product in each order\n",
    "2. We have corresponding dimension tables for customer info, product info, and payment tender info"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:28:01.593980Z",
     "start_time": "2021-12-28T06:28:00.791905Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------+-------------+-------------------+-----------+-------+---+----------+---------------+\n",
      "|order_id|order_item_id|            tran_dt|customer_id|dollars|qty|product_id|payment_type_id|\n",
      "+--------+-------------+-------------------+-----------+-------+---+----------+---------------+\n",
      "|       1|            1|2020-01-01 00:00:00|        572|    550|  1|        20|              2|\n",
      "|       2|            2|2020-01-01 00:00:00|        532|    630|  3|        11|              2|\n",
      "|       3|            3|2020-01-01 00:00:00|        608|    450|  2|        18|              4|\n",
      "|       4|            4|2020-01-01 00:00:00|        424|    110|  2|        10|              2|\n",
      "|       5|            5|2020-01-01 00:00:00|        584|    250|  1|         8|              4|\n",
      "+--------+-------------+-------------------+-----------+-------+---+----------+---------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_sales.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:28:39.842635Z",
     "start_time": "2021-12-28T06:28:03.947843Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(20000, 20000, 19622)"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# order_item_id is the primary key\n",
    "(df_sales.count(),\n",
    " df_sales.selectExpr('count(Distinct order_item_id)').collect()[0][0],\n",
    " df_sales.selectExpr('count(Distinct order_id)').collect()[0][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:28:47.158189Z",
     "start_time": "2021-12-28T06:28:47.116127Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- order_id: integer (nullable = true)\n",
      " |-- order_item_id: integer (nullable = true)\n",
      " |-- tran_dt: timestamp (nullable = true)\n",
      " |-- customer_id: integer (nullable = true)\n",
      " |-- dollars: integer (nullable = true)\n",
      " |-- qty: integer (nullable = true)\n",
      " |-- product_id: integer (nullable = true)\n",
      " |-- payment_type_id: integer (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_sales.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:28:51.881251Z",
     "start_time": "2021-12-28T06:28:51.819430Z"
    }
   },
   "outputs": [],
   "source": [
    "# fix date type for tran_dt\n",
    "df_sales = df_sales.withColumn('tran_dt', F.to_date('tran_dt'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:28:58.146578Z",
     "start_time": "2021-12-28T06:28:56.220554Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+---+---------+------------+----------------+\n",
      "|customer_id|age|hh_income|omni_shopper|email_subscribed|\n",
      "+-----------+---+---------+------------+----------------+\n",
      "|          1| 46|   640000|           0|               0|\n",
      "|          2| 32|   890000|           1|               1|\n",
      "|          3| 45|   772000|           0|               0|\n",
      "|          4| 46|   303000|           0|               1|\n",
      "|          5| 38|   412000|           0|               0|\n",
      "+-----------+---+---------+------------+----------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_customer.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:16.513439Z",
     "start_time": "2021-12-28T06:28:59.254275Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(1000, 1000, 1000)"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# we have 1k unique customers in sales data with all their info in customer dimension table\n",
    "(df_sales.selectExpr('count(Distinct customer_id)').collect()[0][0],\n",
    " df_customer.count(),\n",
    " df_customer.selectExpr('count(Distinct customer_id)').collect()[0][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:18.680933Z",
     "start_time": "2021-12-28T06:29:18.394767Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------+-----+\n",
      "|product_id|category|price|\n",
      "+----------+--------+-----+\n",
      "|         1|       A|  450|\n",
      "|         2|       B|   80|\n",
      "|         3|       C|  250|\n",
      "|         4|       D|  400|\n",
      "|         5|       E|   50|\n",
      "+----------+--------+-----+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# product dimension table provides category and price for each product\n",
    "df_product.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:29.384792Z",
     "start_time": "2021-12-28T06:29:21.334875Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(22, 22)"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "(df_product.count(),\n",
    " df_product.selectExpr('count(Distinct product_id)').collect()[0][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:33.347954Z",
     "start_time": "2021-12-28T06:29:31.585048Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+------------+\n",
      "|payment_type_id|payment_type|\n",
      "+---------------+------------+\n",
      "|              1|        cash|\n",
      "|              2| credit card|\n",
      "|              3|  debit card|\n",
      "|              4|   gift card|\n",
      "|              5|      others|\n",
      "+---------------+------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# payment type table maps the payment type id from sales table\n",
    "df_payment.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Join all dim tables and add week_end column"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:35.011344Z",
     "start_time": "2021-12-28T06:29:34.884601Z"
    }
   },
   "outputs": [],
   "source": [
    "df_sales = df_sales.join(df_product.select('product_id','category'), on=['product_id'], how='left')\n",
    "df_sales = df_sales.join(df_payment, on=['payment_type_id'], how='left')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>week_end column: Saturday of every week</b>\n",
    "\n",
    "`dayofweek()` returns 1-7 corresponding to Sun-Sat for a date.\n",
    "\n",
    "Using this, we will convert each date to the date corresponding to the Saturday of that week (week: Sun-Sat) using below logic:<br/>\n",
    "`date + 7 - dayofweek()`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:39.241910Z",
     "start_time": "2021-12-28T06:29:39.231521Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- payment_type_id: integer (nullable = true)\n",
      " |-- product_id: integer (nullable = true)\n",
      " |-- order_id: integer (nullable = true)\n",
      " |-- order_item_id: integer (nullable = true)\n",
      " |-- tran_dt: date (nullable = true)\n",
      " |-- customer_id: integer (nullable = true)\n",
      " |-- dollars: integer (nullable = true)\n",
      " |-- qty: integer (nullable = true)\n",
      " |-- category: string (nullable = true)\n",
      " |-- payment_type: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_sales.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:42.067028Z",
     "start_time": "2021-12-28T06:29:42.009131Z"
    }
   },
   "outputs": [],
   "source": [
    "df_sales = df_sales.withColumn('week_end',\n",
    "    F.expr(\"date_add(tran_dt, 7-dayofweek(tran_dt))\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:44.277553Z",
     "start_time": "2021-12-28T06:29:42.685492Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------+----------+--------+-------------+----------+-----------+-------+---+--------+------------+----------+\n",
      "|payment_type_id|product_id|order_id|order_item_id|   tran_dt|customer_id|dollars|qty|category|payment_type|  week_end|\n",
      "+---------------+----------+--------+-------------+----------+-----------+-------+---+--------+------------+----------+\n",
      "|              2|        20|       1|            1|2020-01-01|        572|    550|  1|       D| credit card|2020-01-04|\n",
      "|              2|        11|       2|            2|2020-01-01|        532|    630|  3|       A| credit card|2020-01-04|\n",
      "|              4|        18|       3|            3|2020-01-01|        608|    450|  2|       C|   gift card|2020-01-04|\n",
      "|              2|        10|       4|            4|2020-01-01|        424|    110|  2|       E| credit card|2020-01-04|\n",
      "|              4|         8|       5|            5|2020-01-01|        584|    250|  1|       C|   gift card|2020-01-04|\n",
      "+---------------+----------+--------+-------------+----------+-----------+-------+---+--------+------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_sales.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### customer_id x week_end aggregation\n",
    "We will be creating following features at weekly level. These will then be aggregated for multiple time frames using window functions for the final dataset.\n",
    "1. Sales\n",
    "2. No. of orders\n",
    "3. No. of units\n",
    "4. Sales split by category\n",
    "5. Sales split by payment type"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:46.265257Z",
     "start_time": "2021-12-28T06:29:46.164498Z"
    }
   },
   "outputs": [],
   "source": [
    "df_sales_agg = df_sales.groupBy('customer_id','week_end').agg(\n",
    "    F.sum('dollars').alias('sales'),\n",
    "    F.countDistinct('order_id').alias('orders'),\n",
    "    F.sum('qty').alias('units'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:29:53.624832Z",
     "start_time": "2021-12-28T06:29:48.118829Z"
    }
   },
   "outputs": [],
   "source": [
    "# category split pivot\n",
    "df_sales_cat_agg = df_sales.withColumn('category', F.concat(F.lit('cat_'), F.col('category')))\n",
    "\n",
    "df_sales_cat_agg = df_sales_cat_agg.groupBy('customer_id','week_end').pivot('category').agg(F.sum('dollars'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:30:01.000603Z",
     "start_time": "2021-12-28T06:29:56.261970Z"
    }
   },
   "outputs": [],
   "source": [
    "# payment type split pivot\n",
    "# clean-up values in payment type column\n",
    "df_payment_agg = df_sales.withColumn(\n",
    "    'payment_type',\n",
    "    F.concat(F.lit('pay_'), F.regexp_replace(F.col('payment_type'),' ','_')))\n",
    "\n",
    "df_payment_agg = df_payment_agg.groupby('customer_id','week_end').pivot('payment_type').agg(F.max('dollars'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:30:02.100062Z",
     "start_time": "2021-12-28T06:30:01.830172Z"
    }
   },
   "outputs": [],
   "source": [
    "# join all together\n",
    "df_sales_agg = df_sales_agg.join(df_sales_cat_agg, on=['customer_id','week_end'], how='left')\n",
    "df_sales_agg = df_sales_agg.join(df_payment_agg,   on=['customer_id','week_end'], how='left')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:31:56.853869Z",
     "start_time": "2021-12-28T06:30:04.246949Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "17488"
      ]
     },
     "execution_count": 24,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_sales_agg = df_sales_agg.persist()\n",
    "df_sales_agg.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:32:00.843694Z",
     "start_time": "2021-12-28T06:31:59.537386Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+-----+------+-----+-----+-----+-----+-----+-----+--------+---------------+--------------+-------------+----------+\n",
      "|customer_id|  week_end|sales|orders|units|cat_A|cat_B|cat_C|cat_D|cat_E|pay_cash|pay_credit_card|pay_debit_card|pay_gift_card|pay_others|\n",
      "+-----------+----------+-----+------+-----+-----+-----+-----+-----+-----+--------+---------------+--------------+-------------+----------+\n",
      "|         67|2019-10-05| 2300|     2|    5| null| null| null| 2300| null|    null|           1200|          null|         null|      null|\n",
      "|         80|2020-01-11|  900|     1|    2|  900| null| null| null| null|    null|            900|          null|         null|      null|\n",
      "|         81|2020-08-01|  450|     1|    3| null|  450| null| null| null|    null|            450|          null|         null|      null|\n",
      "|         86|2020-02-08|  550|     1|    1|  550| null| null| null| null|    null|            550|          null|         null|      null|\n",
      "|         88|2019-06-01|  740|     2|    2|  450| null|  290| null| null|    null|            450|          null|         null|      null|\n",
      "+-----------+----------+-----+------+-----+-----+-----+-----+-----+-----+--------+---------------+--------------+-------------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_sales_agg.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Fill Missing weeks"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:32:02.503478Z",
     "start_time": "2021-12-28T06:32:02.440040Z"
    }
   },
   "outputs": [],
   "source": [
    "# cust level min and max weeks\n",
    "df_cust = df_sales_agg.groupBy('customer_id').agg(\n",
    "    F.min('week_end').alias('min_week'),\n",
    "    F.max('week_end').alias('max_week'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:32:04.681227Z",
     "start_time": "2021-12-28T06:32:04.665527Z"
    }
   },
   "outputs": [],
   "source": [
    "# function to get a dataframe with 1 row per date in provided range\n",
    "def pandas_date_range(start, end):\n",
    "    dt_rng = pd.date_range(start=start, end=end, freq='W-SAT') # W-SAT required as we want all Saturdays\n",
    "    df_date = pd.DataFrame(dt_rng, columns=['date'])\n",
    "    return df_date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:32:44.964754Z",
     "start_time": "2021-12-28T06:32:06.430312Z"
    }
   },
   "outputs": [],
   "source": [
    "# use the cust level table and create a df with all Saturdays in our range\n",
    "date_list = df_cust.selectExpr('min(min_week)', 'max(max_week)').collect()[0]\n",
    "min_date = date_list[0]\n",
    "max_date = date_list[1]\n",
    "\n",
    "# use the function and create df\n",
    "df_date_range = spark.createDataFrame(pandas_date_range(min_date, max_date))\n",
    "\n",
    "# date format\n",
    "df_date_range = df_date_range.withColumn('date',F.to_date('date'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:32:54.178573Z",
     "start_time": "2021-12-28T06:32:48.026278Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "101"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_date_range = df_date_range.repartition(1).persist()\n",
    "df_date_range.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Cross join date list df with cust table to create filled base table</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:34:09.527209Z",
     "start_time": "2021-12-28T06:34:09.467489Z"
    }
   },
   "outputs": [],
   "source": [
    "df_base = df_cust.crossJoin(F.broadcast(df_date_range))\n",
    "\n",
    "# filter to keep only week_end since first week per customer\n",
    "df_base = df_base.where(F.col('date')>=F.col('min_week'))\n",
    "\n",
    "# rename date to week_end\n",
    "df_base = df_base.withColumnRenamed('date','week_end')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Join with the aggregated week level table to create full base table</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:34:20.803223Z",
     "start_time": "2021-12-28T06:34:20.576040Z"
    }
   },
   "outputs": [],
   "source": [
    "df_base = df_base.join(df_sales_agg, on=['customer_id','week_end'], how='left')\n",
    "df_base = df_base.fillna(0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:35:07.007815Z",
     "start_time": "2021-12-28T06:34:31.259011Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "95197"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_base = df_base.persist()\n",
    "df_base.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:35:27.562602Z",
     "start_time": "2021-12-28T06:35:10.877503Z"
    }
   },
   "outputs": [],
   "source": [
    "# write base table as parquet\n",
    "df_base.repartition(8).write.parquet('./data/rw_base/', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:35:38.906421Z",
     "start_time": "2021-12-28T06:35:38.609396Z"
    }
   },
   "outputs": [],
   "source": [
    "df_base = spark.read.parquet('./data/rw_base/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## y-variable\n",
    "\n",
    "Classification: Determining whether a customer buys something in the next 4 weeks of current week.\n",
    "\n",
    "Regression: Predict total sales in the next 4 weeks of current week."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 35,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:35:47.731756Z",
     "start_time": "2021-12-28T06:35:47.669326Z"
    }
   },
   "outputs": [],
   "source": [
    "# flag 1/0 for weeks with purchases\n",
    "df_base = df_base.withColumn('purchase_flag', F.when(F.col('sales')>0,1).otherwise(0))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:35:51.446528Z",
     "start_time": "2021-12-28T06:35:51.324797Z"
    }
   },
   "outputs": [],
   "source": [
    "# window to aggregate the flag over next 4 weeks\n",
    "df_base = df_base.withColumn(\n",
    "    'purchase_flag_next_4w',\n",
    "    F.max('purchase_flag').over(\n",
    "        Window.partitionBy('customer_id').orderBy('week_end').rowsBetween(1,4)))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [],
   "source": [
    "# window to aggregate the sales over next 4 weeks\n",
    "df_base = df_base.withColumn(\n",
    "    'sales_next_4w',\n",
    "    F.sum('sales').over(\n",
    "        Window.partitionBy('customer_id').orderBy('week_end').rowsBetween(1,4)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Features\n",
    "We will be aggregating the features columns over various time intervals (1/4/13/26/52 weeks) to create a rich set of look-back features. We will also create derived features post aggregation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:06.884253Z",
     "start_time": "2021-12-28T06:36:06.770001Z"
    }
   },
   "outputs": [],
   "source": [
    "# we can create and keep Window() objects that can be referenced in multiple formulas\n",
    "# we don't need a window definition for 1w features as these are already present\n",
    "window_4w  = Window.partitionBy('customer_id').orderBy('week_end').rowsBetween(-3,Window.currentRow)\n",
    "window_13w = Window.partitionBy('customer_id').orderBy('week_end').rowsBetween(-12,Window.currentRow)\n",
    "window_26w = Window.partitionBy('customer_id').orderBy('week_end').rowsBetween(-25,Window.currentRow)\n",
    "window_52w = Window.partitionBy('customer_id').orderBy('week_end').rowsBetween(-51,Window.currentRow)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:13.846492Z",
     "start_time": "2021-12-28T06:36:13.832114Z"
    },
    "scrolled": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['customer_id',\n",
       " 'week_end',\n",
       " 'min_week',\n",
       " 'max_week',\n",
       " 'sales',\n",
       " 'orders',\n",
       " 'units',\n",
       " 'cat_A',\n",
       " 'cat_B',\n",
       " 'cat_C',\n",
       " 'cat_D',\n",
       " 'cat_E',\n",
       " 'pay_cash',\n",
       " 'pay_credit_card',\n",
       " 'pay_debit_card',\n",
       " 'pay_gift_card',\n",
       " 'pay_others',\n",
       " 'purchase_flag',\n",
       " 'purchase_flag_next_4w',\n",
       " 'sales_next_4w']"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_base.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Direct features</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 40,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:25.994456Z",
     "start_time": "2021-12-28T06:36:22.709717Z"
    }
   },
   "outputs": [],
   "source": [
    "cols_skip = ['customer_id','week_end','min_week','max_week','purchase_flag_next_4w','sales_next_4w']\n",
    "for cols in df_base.drop(*cols_skip).columns:\n",
    "    df_base = df_base.withColumn(cols+'_4w',  F.sum(F.col(cols)).over(window_4w))\n",
    "    df_base = df_base.withColumn(cols+'_13w', F.sum(F.col(cols)).over(window_13w))\n",
    "    df_base = df_base.withColumn(cols+'_26w', F.sum(F.col(cols)).over(window_26w))\n",
    "    df_base = df_base.withColumn(cols+'_52w', F.sum(F.col(cols)).over(window_52w))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Derived features</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 41,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:30.080030Z",
     "start_time": "2021-12-28T06:36:27.316223Z"
    }
   },
   "outputs": [],
   "source": [
    "# aov, aur, upt at each time cut\n",
    "for cols in ['sales','orders','units']:\n",
    "    for time_cuts in ['1w','_4w','_13w','_26w','_52w']:\n",
    "        if time_cuts=='1w': time_cuts=''\n",
    "        df_base = df_base.withColumn('aov'+time_cuts, F.col('sales'+time_cuts)/F.col('orders'+time_cuts))\n",
    "        df_base = df_base.withColumn('aur'+time_cuts, F.col('sales'+time_cuts)/F.col('units'+time_cuts))\n",
    "        df_base = df_base.withColumn('upt'+time_cuts, F.col('units'+time_cuts)/F.col('orders'+time_cuts))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 42,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:33.425410Z",
     "start_time": "2021-12-28T06:36:32.766394Z"
    }
   },
   "outputs": [],
   "source": [
    "# % split of category and payment type for 26w (can be extended to other time-frames as well)\n",
    "for cat in ['A','B','C','D','E']:\n",
    "    df_base = df_base.withColumn('cat_'+cat+'_26w_perc', F.col('cat_'+cat+'_26w')/F.col('sales_26w'))\n",
    "\n",
    "for pay in ['cash', 'credit_card', 'debit_card', 'gift_card', 'others']:\n",
    "    df_base = df_base.withColumn('pay_'+pay+'_26w_perc', F.col('pay_'+pay+'_26w')/F.col('sales_26w'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 43,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:34.839166Z",
     "start_time": "2021-12-28T06:36:34.803407Z"
    },
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['customer_id',\n",
       " 'week_end',\n",
       " 'min_week',\n",
       " 'max_week',\n",
       " 'sales',\n",
       " 'orders',\n",
       " 'units',\n",
       " 'cat_A',\n",
       " 'cat_B',\n",
       " 'cat_C',\n",
       " 'cat_D',\n",
       " 'cat_E',\n",
       " 'pay_cash',\n",
       " 'pay_credit_card',\n",
       " 'pay_debit_card',\n",
       " 'pay_gift_card',\n",
       " 'pay_others',\n",
       " 'purchase_flag',\n",
       " 'purchase_flag_next_4w',\n",
       " 'sales_next_4w',\n",
       " 'sales_4w',\n",
       " 'sales_13w',\n",
       " 'sales_26w',\n",
       " 'sales_52w',\n",
       " 'orders_4w',\n",
       " 'orders_13w',\n",
       " 'orders_26w',\n",
       " 'orders_52w',\n",
       " 'units_4w',\n",
       " 'units_13w',\n",
       " 'units_26w',\n",
       " 'units_52w',\n",
       " 'cat_A_4w',\n",
       " 'cat_A_13w',\n",
       " 'cat_A_26w',\n",
       " 'cat_A_52w',\n",
       " 'cat_B_4w',\n",
       " 'cat_B_13w',\n",
       " 'cat_B_26w',\n",
       " 'cat_B_52w',\n",
       " 'cat_C_4w',\n",
       " 'cat_C_13w',\n",
       " 'cat_C_26w',\n",
       " 'cat_C_52w',\n",
       " 'cat_D_4w',\n",
       " 'cat_D_13w',\n",
       " 'cat_D_26w',\n",
       " 'cat_D_52w',\n",
       " 'cat_E_4w',\n",
       " 'cat_E_13w',\n",
       " 'cat_E_26w',\n",
       " 'cat_E_52w',\n",
       " 'pay_cash_4w',\n",
       " 'pay_cash_13w',\n",
       " 'pay_cash_26w',\n",
       " 'pay_cash_52w',\n",
       " 'pay_credit_card_4w',\n",
       " 'pay_credit_card_13w',\n",
       " 'pay_credit_card_26w',\n",
       " 'pay_credit_card_52w',\n",
       " 'pay_debit_card_4w',\n",
       " 'pay_debit_card_13w',\n",
       " 'pay_debit_card_26w',\n",
       " 'pay_debit_card_52w',\n",
       " 'pay_gift_card_4w',\n",
       " 'pay_gift_card_13w',\n",
       " 'pay_gift_card_26w',\n",
       " 'pay_gift_card_52w',\n",
       " 'pay_others_4w',\n",
       " 'pay_others_13w',\n",
       " 'pay_others_26w',\n",
       " 'pay_others_52w',\n",
       " 'purchase_flag_4w',\n",
       " 'purchase_flag_13w',\n",
       " 'purchase_flag_26w',\n",
       " 'purchase_flag_52w',\n",
       " 'aov',\n",
       " 'aur',\n",
       " 'upt',\n",
       " 'aov_4w',\n",
       " 'aur_4w',\n",
       " 'upt_4w',\n",
       " 'aov_13w',\n",
       " 'aur_13w',\n",
       " 'upt_13w',\n",
       " 'aov_26w',\n",
       " 'aur_26w',\n",
       " 'upt_26w',\n",
       " 'aov_52w',\n",
       " 'aur_52w',\n",
       " 'upt_52w',\n",
       " 'cat_A_26w_perc',\n",
       " 'cat_B_26w_perc',\n",
       " 'cat_C_26w_perc',\n",
       " 'cat_D_26w_perc',\n",
       " 'cat_E_26w_perc',\n",
       " 'pay_cash_26w_perc',\n",
       " 'pay_credit_card_26w_perc',\n",
       " 'pay_debit_card_26w_perc',\n",
       " 'pay_gift_card_26w_perc',\n",
       " 'pay_others_26w_perc']"
      ]
     },
     "execution_count": 43,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# all columns\n",
    "df_base.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Derived features: trend vars</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 44,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:55.757327Z",
     "start_time": "2021-12-28T06:36:55.428753Z"
    }
   },
   "outputs": [],
   "source": [
    "# we will take ratio of sales for different time-frames to estimate trend features\n",
    "# that depict whether a customer has an increasing trend or not\n",
    "df_base = df_base.withColumn('sales_1w_over_4w',   F.col('sales')/    F.col('sales_4w'))\n",
    "df_base = df_base.withColumn('sales_4w_over_13w',  F.col('sales_4w')/ F.col('sales_13w'))\n",
    "df_base = df_base.withColumn('sales_13w_over_26w', F.col('sales_13w')/F.col('sales_26w'))\n",
    "df_base = df_base.withColumn('sales_26w_over_52w', F.col('sales_26w')/F.col('sales_52w'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Time elements</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 45,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:36:59.402807Z",
     "start_time": "2021-12-28T06:36:59.224348Z"
    }
   },
   "outputs": [],
   "source": [
    "# extract year, month, and week of year from week_end to be used as features\n",
    "df_base = df_base.withColumn('year', F.year('week_end'))\n",
    "df_base = df_base.withColumn('month', F.month('week_end'))\n",
    "df_base = df_base.withColumn('weekofyear', F.weekofyear('week_end'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>More derived features</b>:<br/>\n",
    "We can add many more derived features as well, as required.\n",
    "\n",
    "e.g. lag variables of existing features, trend ratios for other features, % change (Q-o-Q, M-o-M type) using lag variables, etc."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 46,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:37:25.036807Z",
     "start_time": "2021-12-28T06:37:19.065316Z"
    }
   },
   "outputs": [],
   "source": [
    "# save sample rows to csv for checks\n",
    "df_base.limit(50).toPandas().to_csv('./files/rw_features_qc.csv',index=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 47,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:38:03.016244Z",
     "start_time": "2021-12-28T06:37:27.054620Z"
    }
   },
   "outputs": [],
   "source": [
    "# save features dataset as parquet\n",
    "df_base.repartition(8).write.parquet('./data/rw_features/', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:38:11.095340Z",
     "start_time": "2021-12-28T06:38:10.816881Z"
    }
   },
   "outputs": [],
   "source": [
    "df_features = spark.read.parquet('./data/rw_features/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Model Build"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Dataset for modeling"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Sample one week_end per month</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 48,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:38:21.043453Z",
     "start_time": "2021-12-28T06:38:17.903470Z"
    }
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "24"
      ]
     },
     "execution_count": 48,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_wk_sample = df_features.select('week_end').withColumn('month', F.substring(F.col('week_end'), 1,7))\n",
    "df_wk_sample = df_wk_sample.groupBy('month').agg(F.max('week_end').alias('week_end'))\n",
    "\n",
    "df_wk_sample = df_wk_sample.repartition(1).persist()\n",
    "df_wk_sample.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 49,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:38:27.930926Z",
     "start_time": "2021-12-28T06:38:27.595969Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+----------+\n",
      "|  month|  week_end|\n",
      "+-------+----------+\n",
      "|2019-01|2019-01-26|\n",
      "|2019-02|2019-02-23|\n",
      "|2019-03|2019-03-30|\n",
      "|2019-04|2019-04-27|\n",
      "|2019-05|2019-05-25|\n",
      "+-------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_wk_sample.sort('week_end').show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:38:30.669981Z",
     "start_time": "2021-12-28T06:38:30.010649Z"
    }
   },
   "outputs": [],
   "source": [
    "count_features = df_features.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 51,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:38:41.350445Z",
     "start_time": "2021-12-28T06:38:40.057886Z"
    }
   },
   "outputs": [],
   "source": [
    "# join back to filer\n",
    "df_model = df_features.join(F.broadcast(df_wk_sample.select('week_end')), on=['week_end'], how='inner')\n",
    "count_wk_sample = df_model.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Eligibility filter</b>: Customer should be active in last year w.r.t the reference date"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 52,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:38:58.040892Z",
     "start_time": "2021-12-28T06:38:56.899447Z"
    }
   },
   "outputs": [],
   "source": [
    "# use sales_52w for elig. filter\n",
    "df_model = df_model.where(F.col('sales_52w')>0)\n",
    "count_elig = df_model.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 53,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:39:09.908874Z",
     "start_time": "2021-12-28T06:39:09.891617Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "95197 22938 22938\n"
     ]
    }
   ],
   "source": [
    "# count of rows at each stage\n",
    "print(count_features, count_wk_sample, count_elig)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Removing latest 4 week_end dates</b>: As we have a look-forward period of 4 weeks, latest 4 week_end dates in the data cannot be used for our model as these do not have 4 weeks ahead of them for the y-variable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:39:39.533989Z",
     "start_time": "2021-12-28T06:39:36.342179Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+\n",
      "|  week_end|\n",
      "+----------+\n",
      "|2020-12-05|\n",
      "|2020-11-28|\n",
      "|2020-11-21|\n",
      "|2020-11-14|\n",
      "|2020-11-07|\n",
      "+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# see latest week_end dates (in the dataframe prior to monthly sampling)\n",
    "df_features.select('week_end').drop_duplicates().sort(F.col('week_end').desc()).show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 55,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:39:43.575213Z",
     "start_time": "2021-12-28T06:39:42.379482Z"
    }
   },
   "outputs": [],
   "source": [
    "# filter\n",
    "df_model = df_model.where(F.col('week_end')<'2020-11-14')\n",
    "count_4w_rm = df_model.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {
    "ExecuteTime": {
     "end_time": "2021-12-28T06:39:46.834626Z",
     "start_time": "2021-12-28T06:39:46.822242Z"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "95197 22938 22938 20938\n"
     ]
    }
   ],
   "source": [
    "# count of rows at each stage\n",
    "print(count_features, count_wk_sample, count_elig, count_4w_rm)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Model Dataset Summary\n",
    "Let's look at event rate for our dataset and also get a quick summary of all features.\n",
    "\n",
    "The y-variable is balanced here because it is a dummy dataset. <mark>In most actual scenarios, this will not be balanced and the model build exercise will involving sampling for balancing.</mark>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 57,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+-----+\n",
      "|purchase_flag_next_4w|count|\n",
      "+---------------------+-----+\n",
      "|                    0|10464|\n",
      "|                    1|10474|\n",
      "+---------------------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_model.groupBy('purchase_flag_next_4w').count().sort('purchase_flag_next_4w').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 58,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------+-------------------+\n",
      "|        event_rate|          wk_evt_rt|\n",
      "+------------------+-------------------+\n",
      "|0.5002388002674563|0.18182252364122647|\n",
      "+------------------+-------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_model.groupBy().agg(F.avg('purchase_flag_next_4w').alias('event_rate'), F.avg('purchase_flag').alias('wk_evt_rt')).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Saving summary of all numerical features as a csv</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 59,
   "metadata": {},
   "outputs": [],
   "source": [
    "summary_metrics =\\\n",
    "    ('count','mean','stddev','min','0.10%','1.00%','5.00%','10.00%','20.00%','25.00%','30.00%',\n",
    "     '40.00%','50.00%','60.00%','70.00%','75.00%','80.00%','90.00%','95.00%','99.00%','99.90%','max')\n",
    "\n",
    "df_summary_numeric = df_model.summary(*summary_metrics)\n",
    "df_summary_numeric.toPandas().T.to_csv('./files/rw_features_summary.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 60,
   "metadata": {},
   "outputs": [],
   "source": [
    "# fillna\n",
    "df_model = df_model.fillna(0)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Train-Test Split\n",
    "\n",
    "80-20 split"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 61,
   "metadata": {},
   "outputs": [],
   "source": [
    "train, test = df_model.randomSplit([0.8, 0.2], seed=125)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 62,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['week_end',\n",
       " 'customer_id',\n",
       " 'min_week',\n",
       " 'max_week',\n",
       " 'sales',\n",
       " 'orders',\n",
       " 'units',\n",
       " 'cat_A',\n",
       " 'cat_B',\n",
       " 'cat_C',\n",
       " 'cat_D',\n",
       " 'cat_E',\n",
       " 'pay_cash',\n",
       " 'pay_credit_card',\n",
       " 'pay_debit_card',\n",
       " 'pay_gift_card',\n",
       " 'pay_others',\n",
       " 'purchase_flag',\n",
       " 'purchase_flag_next_4w',\n",
       " 'sales_4w',\n",
       " 'sales_13w',\n",
       " 'sales_26w',\n",
       " 'sales_52w',\n",
       " 'orders_4w',\n",
       " 'orders_13w',\n",
       " 'orders_26w',\n",
       " 'orders_52w',\n",
       " 'units_4w',\n",
       " 'units_13w',\n",
       " 'units_26w',\n",
       " 'units_52w',\n",
       " 'cat_A_4w',\n",
       " 'cat_A_13w',\n",
       " 'cat_A_26w',\n",
       " 'cat_A_52w',\n",
       " 'cat_B_4w',\n",
       " 'cat_B_13w',\n",
       " 'cat_B_26w',\n",
       " 'cat_B_52w',\n",
       " 'cat_C_4w',\n",
       " 'cat_C_13w',\n",
       " 'cat_C_26w',\n",
       " 'cat_C_52w',\n",
       " 'cat_D_4w',\n",
       " 'cat_D_13w',\n",
       " 'cat_D_26w',\n",
       " 'cat_D_52w',\n",
       " 'cat_E_4w',\n",
       " 'cat_E_13w',\n",
       " 'cat_E_26w',\n",
       " 'cat_E_52w',\n",
       " 'pay_cash_4w',\n",
       " 'pay_cash_13w',\n",
       " 'pay_cash_26w',\n",
       " 'pay_cash_52w',\n",
       " 'pay_credit_card_4w',\n",
       " 'pay_credit_card_13w',\n",
       " 'pay_credit_card_26w',\n",
       " 'pay_credit_card_52w',\n",
       " 'pay_debit_card_4w',\n",
       " 'pay_debit_card_13w',\n",
       " 'pay_debit_card_26w',\n",
       " 'pay_debit_card_52w',\n",
       " 'pay_gift_card_4w',\n",
       " 'pay_gift_card_13w',\n",
       " 'pay_gift_card_26w',\n",
       " 'pay_gift_card_52w',\n",
       " 'pay_others_4w',\n",
       " 'pay_others_13w',\n",
       " 'pay_others_26w',\n",
       " 'pay_others_52w',\n",
       " 'purchase_flag_4w',\n",
       " 'purchase_flag_13w',\n",
       " 'purchase_flag_26w',\n",
       " 'purchase_flag_52w',\n",
       " 'aov',\n",
       " 'aur',\n",
       " 'upt',\n",
       " 'aov_4w',\n",
       " 'aur_4w',\n",
       " 'upt_4w',\n",
       " 'aov_13w',\n",
       " 'aur_13w',\n",
       " 'upt_13w',\n",
       " 'aov_26w',\n",
       " 'aur_26w',\n",
       " 'upt_26w',\n",
       " 'aov_52w',\n",
       " 'aur_52w',\n",
       " 'upt_52w',\n",
       " 'cat_A_26w_perc',\n",
       " 'cat_B_26w_perc',\n",
       " 'cat_C_26w_perc',\n",
       " 'cat_D_26w_perc',\n",
       " 'cat_E_26w_perc',\n",
       " 'pay_cash_26w_perc',\n",
       " 'pay_credit_card_26w_perc',\n",
       " 'pay_debit_card_26w_perc',\n",
       " 'pay_gift_card_26w_perc',\n",
       " 'pay_others_26w_perc',\n",
       " 'sales_1w_over_4w',\n",
       " 'sales_4w_over_13w',\n",
       " 'sales_13w_over_26w',\n",
       " 'sales_26w_over_52w',\n",
       " 'year',\n",
       " 'month',\n",
       " 'weekofyear']"
      ]
     },
     "execution_count": 62,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data Prep\n",
    "Spark Models require a vector of features as input. Categorical columns also need to be String Indexed before they can be used.\n",
    "\n",
    "As we don't have any categorical columns currently, we will directly go with VectorAssembler.\n",
    "\n",
    "<b>We will add it to a pipeline model that can be saved to be used on test & scoring datasets.</b>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 63,
   "metadata": {},
   "outputs": [],
   "source": [
    "# model related imports (RF)\n",
    "from pyspark.ml.classification import RandomForestClassifier, RandomForestClassificationModel\n",
    "from pyspark.ml import Pipeline, PipelineModel\n",
    "from pyspark.ml.feature import VectorAssembler, StringIndexer\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 64,
   "metadata": {},
   "outputs": [],
   "source": [
    "# list of features: remove identifier columns and the y-var\n",
    "col_list = df_model.drop('week_end','customer_id','min_week','max_week','purchase_flag_next_4w','sales_next_4w').columns\n",
    "\n",
    "stages = []\n",
    "assembler = VectorAssembler(inputCols=col_list, outputCol='features')\n",
    "stages.append(assembler)\n",
    "\n",
    "pipe = Pipeline(stages=stages)\n",
    "pipe_model = pipe.fit(train)\n",
    "\n",
    "pipe_model.write().overwrite().save('./files/model_objects/rw_pipe/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipe_model = PipelineModel.load('./files/model_objects/rw_pipe/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Apply the transformation pipeline</b>\n",
    "\n",
    "Also keep the identifier columns and y-var in the transformed dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 66,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "16776"
      ]
     },
     "execution_count": 66,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "train_pr = pipe_model.transform(train)\n",
    "train_pr = train_pr.select('customer_id','week_end','purchase_flag_next_4w','features')\n",
    "train_pr = train_pr.persist()\n",
    "train_pr.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 67,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4162"
      ]
     },
     "execution_count": 67,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "test_pr = pipe_model.transform(test)\n",
    "test_pr = test_pr.select('customer_id','week_end','purchase_flag_next_4w','features')\n",
    "test_pr = test_pr.persist()\n",
    "test_pr.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Model Training\n",
    "We will train one iteration of Random Forest model as showcase.\n",
    "\n",
    "In actual scenario, you will have to iterate through the training step multiple times for feature selection, and model hyper parameter tuning to get a good final model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 68,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+---------------------+--------------------+\n",
      "|customer_id|  week_end|purchase_flag_next_4w|            features|\n",
      "+-----------+----------+---------------------+--------------------+\n",
      "|          3|2019-01-26|                    0|(102,[14,15,16,17...|\n",
      "|         14|2019-01-26|                    0|[1200.0,1.0,3.0,0...|\n",
      "|         17|2019-01-26|                    0|(102,[14,15,16,17...|\n",
      "|         19|2019-01-26|                    1|(102,[14,15,16,17...|\n",
      "|         31|2019-01-26|                    1|(102,[0,1,2,6,9,1...|\n",
      "+-----------+----------+---------------------+--------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "train_pr.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 69,
   "metadata": {},
   "outputs": [],
   "source": [
    "model_params = {\n",
    "    'labelCol': 'purchase_flag_next_4w',\n",
    "    'numTrees': 128, # default: 128\n",
    "    'maxDepth': 12,  # default: 12\n",
    "    'featuresCol': 'features',\n",
    "    'minInstancesPerNode': 25,\n",
    "    'maxBins': 128,\n",
    "    'minInfoGain': 0.0,\n",
    "    'subsamplingRate': 0.7,\n",
    "    'featureSubsetStrategy': '0.3',\n",
    "    'impurity': 'gini',\n",
    "    'seed': 125,\n",
    "    'cacheNodeIds': False,\n",
    "    'maxMemoryInMB': 256\n",
    "    }\n",
    "\n",
    "clf = RandomForestClassifier(**model_params)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [],
   "source": [
    "trained_clf = clf.fit(train_pr)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Feature Importance\n",
    "We will save feature importance as a csv."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Feature importance\n",
    "feature_importance_list = trained_clf.featureImportances\n",
    "feature_list = pd.DataFrame(train_pr.schema['features'].metadata['ml_attr']['attrs']['numeric']).sort_values('idx')\n",
    "\n",
    "feature_importance_list = pd.DataFrame(\n",
    "    data=feature_importance_list.toArray(),\n",
    "    columns=['relative_importance'],\n",
    "    index=feature_list['name'])\n",
    "feature_importance_list = feature_importance_list.sort_values('relative_importance', ascending=False)\n",
    "\n",
    "feature_importance_list.to_csv('./files/rw_rf_feat_imp.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Predict on train and test"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [],
   "source": [
    "secondelement = F.udf(lambda v: float(v[1]), FloatType())\n",
    "\n",
    "train_pred = trained_clf.transform(train_pr).withColumn('score',secondelement(F.col('probability')))\n",
    "test_pred =  trained_clf.transform(test_pr).withColumn('score', secondelement(F.col('probability')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+---------------------+--------------------+--------------------+--------------------+----------+----------+\n",
      "|customer_id|  week_end|purchase_flag_next_4w|            features|       rawPrediction|         probability|prediction|     score|\n",
      "+-----------+----------+---------------------+--------------------+--------------------+--------------------+----------+----------+\n",
      "|         15|2019-01-26|                    1|(102,[0,1,2,6,9,1...|[53.0784863119179...|[0.41467567431185...|       1.0|0.58532435|\n",
      "|         27|2019-01-26|                    1|(102,[14,15,16,17...|[51.1393971441854...|[0.39952654018894...|       1.0|0.60047346|\n",
      "|         28|2019-01-26|                    1|(102,[14,15,16,17...|[49.8725279055357...|[0.38962912426199...|       1.0| 0.6103709|\n",
      "|        170|2019-01-26|                    0|(102,[0,1,2,7,10,...|[54.1789500442362...|[0.42327304722059...|       1.0|  0.576727|\n",
      "|        192|2019-01-26|                    0|(102,[0,1,2,7,10,...|[57.5312187445332...|[0.44946264644166...|       1.0|0.55053735|\n",
      "+-----------+----------+---------------------+--------------------+--------------------+--------------------+----------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "test_pred.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test Set Evaluation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 74,
   "metadata": {},
   "outputs": [],
   "source": [
    "evaluator = BinaryClassificationEvaluator(\n",
    "        rawPredictionCol='rawPrediction',\n",
    "        labelCol='purchase_flag_next_4w',\n",
    "        metricName='areaUnderROC')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 75,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.8116811886255015"
      ]
     },
     "execution_count": 75,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# areaUnderROC\n",
    "evaluator.evaluate(train_pred)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 76,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.7412597272276923"
      ]
     },
     "execution_count": 76,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "evaluator.evaluate(test_pred)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+----------+-----+\n",
      "|purchase_flag_next_4w|prediction|count|\n",
      "+---------------------+----------+-----+\n",
      "|                    0|       0.0| 1655|\n",
      "|                    0|       1.0|  450|\n",
      "|                    1|       0.0|  937|\n",
      "|                    1|       1.0| 1120|\n",
      "+---------------------+----------+-----+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# cm\n",
    "test_pred.groupBy('purchase_flag_next_4w','prediction').count().sort('purchase_flag_next_4w','prediction').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 78,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "0.6667467563671312"
      ]
     },
     "execution_count": 78,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# accuracy\n",
    "test_pred.where(F.col('purchase_flag_next_4w')==F.col('prediction')).count()/test_pred.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Save Model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 79,
   "metadata": {},
   "outputs": [],
   "source": [
    "trained_clf.write().overwrite().save('./files/model_objects/rw_rf_model/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [],
   "source": [
    "trained_clf = RandomForestClassificationModel.load('./files/model_objects/rw_rf_model/')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Scoring\n",
    "We will take the records for latest week_end from df_features and score it using our trained model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 81,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_features = spark.read.parquet('./data/rw_features/')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "datetime.date(2020, 12, 5)"
      ]
     },
     "execution_count": 82,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "max_we = df_features.selectExpr('max(week_end)').collect()[0][0]\n",
    "max_we"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 88,
   "metadata": {},
   "outputs": [],
   "source": [
    "df_scoring = df_features.where(F.col('week_end')==max_we)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1000"
      ]
     },
     "execution_count": 89,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_scoring.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 90,
   "metadata": {},
   "outputs": [],
   "source": [
    "# fillna\n",
    "df_scoring = df_scoring.fillna(0)\n",
    "\n",
    "# transformation pipeline\n",
    "pipe_model = PipelineModel.load('./files/model_objects/rw_pipe/')\n",
    "\n",
    "# apply\n",
    "df_scoring = pipe_model.transform(df_scoring)\n",
    "df_scoring = df_scoring.select('customer_id','week_end','features')\n",
    "\n",
    "# rf model\n",
    "trained_clf = RandomForestClassificationModel.load('./files/model_objects/rw_rf_model/')\n",
    "\n",
    "#apply\n",
    "secondelement = F.udf(lambda v: float(v[1]), FloatType())\n",
    "\n",
    "df_scoring = trained_clf.transform(df_scoring).withColumn('score',secondelement(F.col('probability')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 91,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+----------+--------------------+--------------------+--------------------+----------+----------+\n",
      "|customer_id|  week_end|            features|       rawPrediction|         probability|prediction|     score|\n",
      "+-----------+----------+--------------------+--------------------+--------------------+----------+----------+\n",
      "|        148|2020-12-05|[1200.0,1.0,3.0,1...|[68.4381452599104...|[0.53467300984305...|       0.0|  0.465327|\n",
      "|        787|2020-12-05|(102,[15,16,17,19...|[96.4135634068059...|[0.75323096411567...|       0.0|0.24676904|\n",
      "|        906|2020-12-05|(102,[16,17,20,21...|[90.4284145792717...|[0.70647198890056...|       0.0|0.29352802|\n",
      "|        182|2020-12-05|(102,[16,17,20,21...|[90.9264612678324...|[0.71036297865494...|       0.0|0.28963703|\n",
      "|        442|2020-12-05|(102,[16,17,20,21...|[105.838704191258...|[0.82686487649421...|       0.0|0.17313512|\n",
      "+-----------+----------+--------------------+--------------------+--------------------+----------+----------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_scoring.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 92,
   "metadata": {},
   "outputs": [],
   "source": [
    "# save scored output\n",
    "df_scoring.repartition(8).write.parquet('./data/rw_scored/', mode='overwrite')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.7.9"
  },
  "pycharm": {
   "stem_cell": {
    "cell_type": "raw",
    "metadata": {
     "collapsed": false
    },
    "source": []
   }
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": true,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": true,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "368px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}