SparkSQL.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkContext, SparkConf\n",
    "from pyspark.sql import SparkSession\n",
    "from pyspark.sql.types import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext.getOrCreate()\n",
    "ss = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "instacart = ss.read.csv('/Users/christopherolley/data/consolidated_df.csv', header=True, inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- _c0: integer (nullable = true)\n",
      " |-- order_id: integer (nullable = true)\n",
      " |-- user_id: integer (nullable = true)\n",
      " |-- eval_set: string (nullable = true)\n",
      " |-- order_number: integer (nullable = true)\n",
      " |-- order_dow: integer (nullable = true)\n",
      " |-- order_hour_of_day: integer (nullable = true)\n",
      " |-- days_since_prior_order: double (nullable = true)\n",
      " |-- product_id: double (nullable = true)\n",
      " |-- add_to_cart_order: double (nullable = true)\n",
      " |-- reordered: double (nullable = true)\n",
      " |-- product_name: string (nullable = true)\n",
      " |-- aisle_id: string (nullable = true)\n",
      " |-- department_id: string (nullable = true)\n",
      " |-- department: string (nullable = true)\n",
      " |-- aisle: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "instacart.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "instacart.write.saveAsTable('Instacart')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Product Features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Number of times a product was ordered:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+------+\n",
      "|product_id| count|\n",
      "+----------+------+\n",
      "|   24852.0|472565|\n",
      "|   13176.0|379450|\n",
      "|   21137.0|264683|\n",
      "|   21903.0|241921|\n",
      "|   47209.0|213584|\n",
      "|   47766.0|176815|\n",
      "|   47626.0|152657|\n",
      "|   16797.0|142951|\n",
      "|   26209.0|140627|\n",
      "|   27845.0|137905|\n",
      "|   27966.0|137057|\n",
      "|   22935.0|113426|\n",
      "|   24964.0|109778|\n",
      "|   45007.0|104823|\n",
      "|   39275.0|100060|\n",
      "|   49683.0| 97315|\n",
      "|   28204.0| 89632|\n",
      "|    5876.0| 87746|\n",
      "|    8277.0| 85020|\n",
      "|   40706.0| 84255|\n",
      "+----------+------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "instacart.filter(\"eval_set=='prior'\").groupBy('product_id').count().orderBy(\"count\", ascending=False).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Average position in cart:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+--------------------+----------------------+\n",
      "|product_id|        product_name|avg(add_to_cart_order)|\n",
      "+----------+--------------------+----------------------+\n",
      "|   37088.0|Lndbrg White Quin...|                   1.0|\n",
      "|   39759.0|Indoor & Outdoor ...|                   1.0|\n",
      "|   44495.0|Lowfat Cherry Lim...|                   1.0|\n",
      "|   25441.0|       Spicy Falafel|                   1.0|\n",
      "|   41800.0|       Easter Basket|                   1.0|\n",
      "|   43144.0|Flax Oil, Organic...|                   1.0|\n",
      "|   33440.0|Rosa Mosqueta Ros...|                   1.0|\n",
      "|    2769.0|Pappardelle Nests...|                   1.0|\n",
      "|   12841.0|      King Crab Legs|                   1.0|\n",
      "|   34962.0|  Prenatal Nutrients|                   1.0|\n",
      "|   25888.0|Super Nutrition V...|                   1.0|\n",
      "|   27863.0|        Cooking Fuel|                   1.0|\n",
      "|   47680.0| Indian Wells Merlot|                   1.0|\n",
      "|    2796.0|      American Blend|                   1.0|\n",
      "|   36597.0|Vanilla Flavor Mu...|                   1.0|\n",
      "|   38276.0|Chocolate Pepperm...|                   1.0|\n",
      "|   11254.0|Drink Distinct Al...|                   1.0|\n",
      "|   34463.0|Vegan Coconut Mil...|                   1.0|\n",
      "|   30343.0|Seltzer Blueberry...|                   1.0|\n",
      "|   41790.0|Peppermint Schnap...|                   1.0|\n",
      "+----------+--------------------+----------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "instacart.filter(\"eval_set=='prior'\").groupBy('product_id', 'product_name').avg(\"add_to_cart_order\").orderBy(\"avg(add_to_cart_order)\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Reorder rate: number of reorders/total number of times ordered"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# User x Product Features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Reorder count: number of time each user has reordered a product:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+--------------+\n",
      "|product_id|user_id|sum(reordered)|\n",
      "+----------+-------+--------------+\n",
      "|   35547.0|    385|           0.0|\n",
      "|   27845.0|    433|           1.0|\n",
      "|   34050.0|    601|           4.0|\n",
      "|   44949.0|    853|           9.0|\n",
      "|   35221.0|   1123|           0.0|\n",
      "|   13113.0|   1607|           2.0|\n",
      "|   18224.0|   2187|           3.0|\n",
      "|   30620.0|   2451|           0.0|\n",
      "|    5612.0|   2736|           7.0|\n",
      "|   33198.0|   2954|           0.0|\n",
      "|   24852.0|   2991|           6.0|\n",
      "|   15424.0|   3686|           2.0|\n",
      "|   30779.0|   3781|           1.0|\n",
      "|   45989.0|   4381|          10.0|\n",
      "|   14705.0|   4630|           1.0|\n",
      "|   35547.0|   5104|           4.0|\n",
      "|   38387.0|   5117|           2.0|\n",
      "|   37687.0|   5245|           2.0|\n",
      "|    4987.0|   5387|           4.0|\n",
      "|    6377.0|   5470|           1.0|\n",
      "+----------+-------+--------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "instacart.filter(\"eval_set=='prior'\").groupBy('product_id', 'user_id').sum('reordered').show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Number of orders since a user last ordered a given item: done by generating chrononological order_num from order_id, and returns max order_num (grouped by user) - max order_num (grouped by user and product)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 99,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+------------------------------+\n",
      "|product_id|user_id|num_since_last_order_user_item|\n",
      "+----------+-------+------------------------------+\n",
      "|     132.0|    148|                             1|\n",
      "|     274.0|    148|                             4|\n",
      "|     781.0|    148|                             3|\n",
      "|    1203.0|    148|                             1|\n",
      "|    1915.0|    148|                             1|\n",
      "|    3376.0|    148|                             3|\n",
      "|    4025.0|    148|                             5|\n",
      "|    4477.0|    148|                             5|\n",
      "|    5201.0|    148|                             1|\n",
      "|    6891.0|    148|                             1|\n",
      "|    7412.0|    148|                             1|\n",
      "|    7539.0|    148|                             1|\n",
      "|    7870.0|    148|                             1|\n",
      "|    8362.0|    148|                             3|\n",
      "|   10129.0|    148|                             0|\n",
      "|   10451.0|    148|                             6|\n",
      "|   10983.0|    148|                             1|\n",
      "|   11109.0|    148|                             1|\n",
      "|   11212.0|    148|                             1|\n",
      "|   12023.0|    148|                             1|\n",
      "|   13577.0|    148|                             1|\n",
      "|   14804.0|    148|                             5|\n",
      "|   15008.0|    148|                             1|\n",
      "|   15438.0|    148|                             6|\n",
      "|   17284.0|    148|                             0|\n",
      "|   17450.0|    148|                             1|\n",
      "|   17828.0|    148|                             1|\n",
      "|   18629.0|    148|                             1|\n",
      "|   19385.0|    148|                             5|\n",
      "|   20310.0|    148|                             1|\n",
      "|   20564.0|    148|                             1|\n",
      "|   20632.0|    148|                             1|\n",
      "|   21295.0|    148|                             1|\n",
      "|   21806.0|    148|                             1|\n",
      "|   22035.0|    148|                             6|\n",
      "|   22217.0|    148|                             5|\n",
      "|   22721.0|    148|                             1|\n",
      "|   23383.0|    148|                             1|\n",
      "|   23459.0|    148|                             1|\n",
      "|   24118.0|    148|                             5|\n",
      "|   24230.0|    148|                             0|\n",
      "|   24413.0|    148|                             1|\n",
      "|   25821.0|    148|                             6|\n",
      "|   26313.0|    148|                             5|\n",
      "|   26638.0|    148|                             5|\n",
      "|   27156.0|    148|                             3|\n",
      "|   27219.0|    148|                             5|\n",
      "|   27845.0|    148|                             1|\n",
      "|   28568.0|    148|                             5|\n",
      "|   29406.0|    148|                             3|\n",
      "+----------+-------+------------------------------+\n",
      "only showing top 50 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ss.sql(\"select distinct product_id, user_id,\\\n",
    "        max(order_num) over (partition by user_id) - max(order_num) over (partition by user_id, product_id) as num_since_last_order_user_item from\\\n",
    "        (select Instacart.order_id, Instacart.user_id, Instacart.product_id, rhs.order_num\\\n",
    "        from Instacart\\\n",
    "        left join\\\n",
    "        (select order_id, user_id, row_number() over (partition by user_id order by order_id) as order_num from\\\n",
    "        (select distinct order_id, user_id from Instacart where eval_set = 'prior') as iq) as rhs\\\n",
    "        on Instacart.order_id=rhs.order_id and Instacart.user_id=rhs.user_id\\\n",
    "        where eval_set = 'prior') as iq2\").show(50)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Rate of user item reorder: # of reorders of an item / # of orders since first time ordering item.\n",
    "Get max(order_num) grouped by user_id, then min(order_num) grouped by user_id and product, subtract the two to get number of orders since first purchase of an item. Then sum(reordered) grouped by item, user to get the number of times an item was reordered by a user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 108,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------+-------+-------------------------+------------------------------+-------------------------+\n",
      "|product_id|user_id|num_reorders_product_user|since_first_order_user_product|reorder_rate_product_user|\n",
      "+----------+-------+-------------------------+------------------------------+-------------------------+\n",
      "|     132.0|    148|                      0.0|                             1|                      0.0|\n",
      "|     274.0|    148|                      0.0|                             4|                      0.0|\n",
      "|     781.0|    148|                      0.0|                             3|                      0.0|\n",
      "|    1203.0|    148|                      0.0|                             1|                      0.0|\n",
      "|    1915.0|    148|                      0.0|                             1|                      0.0|\n",
      "|    3376.0|    148|                      0.0|                             3|                      0.0|\n",
      "|    4025.0|    148|                      0.0|                             5|                      0.0|\n",
      "|    4477.0|    148|                      0.0|                             5|                      0.0|\n",
      "|    5201.0|    148|                      0.0|                             1|                      0.0|\n",
      "|    6891.0|    148|                      0.0|                             1|                      0.0|\n",
      "|    7412.0|    148|                      1.0|                             6|      0.16666666666666666|\n",
      "|    7539.0|    148|                      0.0|                             1|                      0.0|\n",
      "|    7870.0|    148|                      1.0|                             4|                     0.25|\n",
      "|    8362.0|    148|                      0.0|                             3|                      0.0|\n",
      "|   10129.0|    148|                      0.0|                             0|                     null|\n",
      "|   10451.0|    148|                      0.0|                             6|                      0.0|\n",
      "|   10983.0|    148|                      1.0|                             3|       0.3333333333333333|\n",
      "|   11109.0|    148|                      0.0|                             1|                      0.0|\n",
      "|   11212.0|    148|                      0.0|                             1|                      0.0|\n",
      "|   12023.0|    148|                      0.0|                             1|                      0.0|\n",
      "+----------+-------+-------------------------+------------------------------+-------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ss.sql(\"select product_id, user_id, num_reorders_product_user, since_first_order_user_product,\\\n",
    "        num_reorders_product_user/since_first_order_user_product as reorder_rate_product_user from\\\n",
    "        (select distinct product_id, user_id,\\\n",
    "        max(order_num) over (partition by user_id) - min(order_num) over (partition by user_id, product_id) as since_first_order_user_product,\\\n",
    "        sum(reordered) over (partition by user_id, product_id) as num_reorders_product_user from\\\n",
    "        (select Instacart.order_id, Instacart.user_id, Instacart.product_id, Instacart.reordered, rhs.order_num\\\n",
    "        from Instacart\\\n",
    "        left join\\\n",
    "        (select order_id, user_id, row_number() over (partition by user_id order by order_id) as order_num from\\\n",
    "        (select distinct order_id, user_id from Instacart where eval_set = 'prior') as iq) as rhs\\\n",
    "        on Instacart.order_id=rhs.order_id and Instacart.user_id=rhs.user_id\\\n",
    "        where eval_set = 'prior') as iq2) as iq3\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# User Features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Maximum number of times a user has ordered any product:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 54,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------+\n",
      "|   uid|max(ct)|\n",
      "+------+-------+\n",
      "| 11141|      2|\n",
      "| 46266|     10|\n",
      "| 16386|     27|\n",
      "|160492|     10|\n",
      "|156366|     33|\n",
      "| 11748|      4|\n",
      "| 17679|     56|\n",
      "|  7993|      7|\n",
      "| 63271|     34|\n",
      "|156197|      6|\n",
      "| 57380|     14|\n",
      "|  3794|     38|\n",
      "|  4900|     13|\n",
      "| 43103|      4|\n",
      "| 43714|     16|\n",
      "|  9376|     20|\n",
      "|156749|      4|\n",
      "| 43527|     30|\n",
      "| 43302|      8|\n",
      "|160563|      7|\n",
      "+------+-------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ss.sql(\"select uid, max(ct) from\\\n",
    "       (select product_id as pid, user_id as uid, count(*) as ct from Instacart\\\n",
    "       where eval_set = 'prior'\\\n",
    "       group by 1,2) as iq\\\n",
    "       group by 1\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Average order size by user:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 77,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------+\n",
      "|user_id|   avg(order_size)|\n",
      "+-------+------------------+\n",
      "|    496| 5.512195121951219|\n",
      "|   4101|              18.4|\n",
      "|   5156|               9.3|\n",
      "|   6654|              10.0|\n",
      "|   9376| 12.13888888888889|\n",
      "|  18800|               3.0|\n",
      "|  19984|12.785714285714286|\n",
      "|  20924|         10.046875|\n",
      "|  44906|12.941176470588236|\n",
      "|  46994| 7.029411764705882|\n",
      "|  57370| 9.619047619047619|\n",
      "|  60769|18.413793103448278|\n",
      "|  63106|15.885714285714286|\n",
      "| 155959| 6.333333333333333|\n",
      "| 156366|13.886075949367088|\n",
      "| 160563|14.909090909090908|\n",
      "| 133189|14.909090909090908|\n",
      "| 162321|11.181818181818182|\n",
      "| 164962| 4.925925925925926|\n",
      "| 167316| 6.777777777777778|\n",
      "+-------+------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ss.sql(\"select user_id, avg(order_size) from\\\n",
    "        (select user_id, order_id, count(*) as order_size from Instacart\\\n",
    "        where eval_set = 'prior'\\\n",
    "        group by 1,2) as iq\\\n",
    "        group by 1\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Number of aisles purchased from:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 80,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+---------------------+\n",
      "|user_id|count(DISTINCT aisle)|\n",
      "+-------+---------------------+\n",
      "|   2659|                   59|\n",
      "|   2866|                   19|\n",
      "|   3794|                   64|\n",
      "|   3918|                   40|\n",
      "|   5518|                   27|\n",
      "|   7554|                   41|\n",
      "|  13289|                   45|\n",
      "|  15447|                   27|\n",
      "|  19553|                    7|\n",
      "|  20924|                   66|\n",
      "|  47283|                   58|\n",
      "|  48254|                   52|\n",
      "|  56640|                   35|\n",
      "|  59355|                   31|\n",
      "|  61051|                   16|\n",
      "|  63155|                   36|\n",
      "| 155042|                   81|\n",
      "| 156527|                    9|\n",
      "| 158257|                   45|\n",
      "| 160767|                   21|\n",
      "+-------+---------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ss.sql(\"select user_id, count(distinct aisle) from Instacart\\\n",
    "        where eval_set = 'prior'\\\n",
    "        group by 1\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Number of departments purchased from:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 82,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+--------------------------+\n",
      "|user_id|count(DISTINCT department)|\n",
      "+-------+--------------------------+\n",
      "|    148|                        15|\n",
      "|   1591|                         9|\n",
      "|   6620|                        11|\n",
      "|  12799|                         7|\n",
      "|  17420|                        15|\n",
      "|  18024|                         7|\n",
      "|  18051|                        10|\n",
      "|  19079|                         8|\n",
      "|  21220|                        15|\n",
      "|  43935|                        15|\n",
      "|  57039|                        11|\n",
      "| 156296|                         9|\n",
      "| 158593|                        12|\n",
      "| 128389|                        14|\n",
      "| 129153|                         7|\n",
      "| 114851|                        19|\n",
      "|  77803|                        13|\n",
      "|  85749|                        10|\n",
      "|  89476|                         9|\n",
      "|  90019|                        14|\n",
      "+-------+--------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ss.sql(\"select user_id, count(distinct department) from Instacart where eval_set = 'prior'\\\n",
    "        group by 1\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Number of orders made by each user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 103,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------------+\n",
      "|user_id|count(DISTINCT order_id)|\n",
      "+-------+------------------------+\n",
      "|    496|                      82|\n",
      "|   4101|                       5|\n",
      "|   5156|                      30|\n",
      "|   6654|                       9|\n",
      "|   9376|                      36|\n",
      "|  18800|                       9|\n",
      "|  19984|                      14|\n",
      "|  20924|                      64|\n",
      "|  44906|                      17|\n",
      "|  46994|                      34|\n",
      "|  57370|                      21|\n",
      "|  60769|                      29|\n",
      "|  63106|                      35|\n",
      "| 155959|                      12|\n",
      "| 156366|                      79|\n",
      "| 160563|                      11|\n",
      "| 133189|                      11|\n",
      "| 162321|                      22|\n",
      "| 164962|                      27|\n",
      "| 167316|                      99|\n",
      "+-------+------------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ss.sql(\"select user_id, count(distinct order_id) from Instacart\\\n",
    "        where eval_set = 'prior'\\\n",
    "        group by 1\").show()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}