Lab 3 - Lecture.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Lab 3 - Spark MLlib\n",
    "\n",
    "##### \"A computer program is said to learn from experience E with respect to some class of tasks T and performance measure P if its performance at tasks in T, as measured by P, improves with experience E\"\n",
    "-Tom M. Mitchell\n",
    "\n",
    "#### Machine Learning - the science of getting computers to act without being explicitly programmed\n",
    "\n",
    "MLlib is Spark’s machine learning (ML) library. Its goal is to make practical machine learning scalable and easy. It consists of common learning algorithms and utilities, including classification, regression, clustering, collaborative filtering (this example!), dimensionality reduction, as well as lower-level optimization primitives and higher-level pipeline APIs.\n",
    "\n",
    "It divides into two packages:\n",
    "- spark.mllib contains the original API built on top of RDDs.\n",
    "- spark.ml provides higher-level API built on top of DataFrames for constructing ML pipelines.\n",
    "\n",
    "\n",
    "Using spark.ml is recommended because with DataFrames the API is more versatile and flexible. But we will keep supporting spark.mllib along with the development of spark.ml. Users should be comfortable using spark.mllib features and expect more features coming.\n",
    "\n",
    "http://spark.apache.org/docs/latest/mllib-guide.html\n",
    "\n",
    "## Online Purchase Recommendations\n",
    "\n",
    "Learn how to create a recommendation engine using the Alternating Least Squares algorithm in Spark's machine learning library\n",
    "\n",
    "<img src='https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/ALS.png' width=\"70%\" height=\"70%\"></img>\n",
    "\n",
    "### The data\n",
    "\n",
    "This is a transnational data set which contains all the transactions occurring between 01/12/2010 and 09/12/2011 for a UK-based and registered non-store online retail.  The company mainly sells unique all-occasion gifts. Many customers of the company are wholesalers.\n",
    "\n",
    "http://archive.ics.uci.edu/ml/datasets/Online+Retail\n",
    "\n",
    "<img src='https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/FullFile.png' width=\"80%\" height=\"80%\"></img>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Create an RDD from the csv data "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Download the data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--2016-06-21 10:35:40--  https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/OnlineRetail.csv.gz\n",
      "Resolving raw.githubusercontent.com (raw.githubusercontent.com)... 23.235.40.133\n",
      "Connecting to raw.githubusercontent.com (raw.githubusercontent.com)|23.235.40.133|:443... connected.\n",
      "HTTP request sent, awaiting response... 200 OK\n",
      "Length: 7483128 (7.1M) [application/octet-stream]\n",
      "Saving to: 'OnlineRetail.csv.gz'\n",
      "\n",
      "100%[======================================>] 7,483,128   12.1MB/s   in 0.6s   \n",
      "\n",
      "2016-06-21 10:35:41 (12.1 MB/s) - 'OnlineRetail.csv.gz' saved [7483128/7483128]\n",
      "\n"
     ]
    }
   ],
   "source": [
    "!rm 'OnlineRetail.csv.gz' -f\n",
    "!wget https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/OnlineRetail.csv.gz"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Put the csv into an RDD (at first, each row in the RDD is a string which correlates to a line in the csv)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[u'InvoiceNo,StockCode,Description,Quantity,InvoiceDate,UnitPrice,CustomerID,Country', u'536365,85123A,WHITE HANGING HEART T-LIGHT HOLDER,6,12/1/10 8:26,2.55,17850,United Kingdom']\n"
     ]
    }
   ],
   "source": [
    "loadRetailData = sc.textFile(\"OnlineRetail.csv.gz\")\n",
    "print loadRetailData.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prepare and shape the data:  \"80% of a Data Scientists  job\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Remove the header from the RDD and split the string in each row by comma"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[u'536365', u'85123A', u'WHITE HANGING HEART T-LIGHT HOLDER', u'6', u'12/1/10 8:26', u'2.55', u'17850', u'United Kingdom'], [u'536365', u'71053', u'WHITE METAL LANTERN', u'6', u'12/1/10 8:26', u'3.39', u'17850', u'United Kingdom']]\n"
     ]
    }
   ],
   "source": [
    "header = loadRetailData.first()\n",
    "loadRetailData = loadRetailData.filter(lambda line: line != header).\\\n",
    "                            map(lambda l: l.split(\",\"))\n",
    "\n",
    "print loadRetailData.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### NOTE:  The original file at UCI's Machine Learning Repository has commas in the product description.  Those have been removed to expediate the lab.\n",
    "#### Only keep rows that have a purchase quantity of greater than 0, a customerID not equal to 0, and a non blank stock code after removing non-numeric characters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[[u'536365', u'85123A', u'WHITE HANGING HEART T-LIGHT HOLDER', u'6', u'12/1/10 8:26', u'2.55', u'17850', u'United Kingdom'], [u'536365', u'71053', u'WHITE METAL LANTERN', u'6', u'12/1/10 8:26', u'3.39', u'17850', u'United Kingdom']]\n"
     ]
    }
   ],
   "source": [
    "import re\n",
    "\n",
    "loadRetailData = loadRetailData.filter(lambda l: int(l[3]) > 0\\\n",
    "                                and len(re.sub(\"\\D\", \"\", l[1])) != 0 \\\n",
    "                                and len(l[6]) != 0)\n",
    "\n",
    "print loadRetailData.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Map each line to a row and create a data frame "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- country: string (nullable = true)\n",
      " |-- custId: long (nullable = true)\n",
      " |-- description: string (nullable = true)\n",
      " |-- inv: long (nullable = true)\n",
      " |-- invDate: string (nullable = true)\n",
      " |-- price: double (nullable = true)\n",
      " |-- quant: long (nullable = true)\n",
      " |-- stockCode: long (nullable = true)\n",
      "\n",
      "None\n",
      "          country  custId                         description     inv  \\\n",
      "0  United Kingdom   17850  WHITE HANGING HEART T-LIGHT HOLDER  536365   \n",
      "1  United Kingdom   17850                 WHITE METAL LANTERN  536365   \n",
      "\n",
      "        invDate  price  quant  stockCode  \n",
      "0  12/1/10 8:26   2.55      6      85123  \n",
      "1  12/1/10 8:26   3.39      6      71053  \n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql import SQLContext, Row\n",
    "sqlContext = SQLContext(sc)\n",
    "\n",
    "#Convert each line to a Row.\n",
    "loadRetailData = loadRetailData.map(lambda l: Row(inv=int(l[0]),\\\n",
    "                                    stockCode=int(re.sub(\"\\D\", \"\", l[1])),\\\n",
    "                                    description=l[2],\\\n",
    "                                    quant=int(l[3]),\\\n",
    "                                    invDate=l[4],\\\n",
    "                                    price=float(l[5]),\\\n",
    "                                    custId=int(l[6]),\\\n",
    "                                    country=l[7]))\n",
    "\n",
    "# Infer the schema, and register the DataFrame as a table.\n",
    "retailDf = sqlContext.createDataFrame(loadRetailData)\n",
    "print retailDf.printSchema()\n",
    "\n",
    "retailDf.registerTempTable(\"retailPurchases\")\n",
    "print sqlContext.sql(\"SELECT * FROM retailPurchases limit 2\").toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Keep only the data we need (custId, stockCode, and rank)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(custId=12838, stockCode=22941, purch=1), Row(custId=17968, stockCode=22731, purch=1)]\n"
     ]
    }
   ],
   "source": [
    "query = \"\"\"\n",
    "SELECT \n",
    "    custId, stockCode, 1 as purch\n",
    "FROM \n",
    "    retailPurchases \n",
    "group \n",
    "    by custId, stockCode\"\"\"\n",
    "retailDf = sqlContext.sql(query)\n",
    "\n",
    "print retailDf.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Randomly split the data into a testing set (10% of the data), a cross validation set (10% of the data) a training set (80% of the data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(custId=17968, stockCode=22731, purch=1), Row(custId=17897, stockCode=84558, purch=1)]\n",
      "[Row(custId=12838, stockCode=22941, purch=1), Row(custId=13468, stockCode=21231, purch=1)]\n",
      "[Row(custId=16210, stockCode=20977, purch=1), Row(custId=13090, stockCode=22617, purch=1)]\n"
     ]
    }
   ],
   "source": [
    "testDf, cvDf, trainDf = retailDf.randomSplit([.1,.1,.8],1)\n",
    "\n",
    "print trainDf.take(2)\n",
    "print cvDf.take(2)\n",
    "print testDf.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Build recommendation models\n",
    "\n",
    "#### Use training DF to train a model with Alternating Least Squares \n",
    "Latent Factors / rank<br>\n",
    "The number of columns in the user-feature and product-feature matricies)<br>\n",
    "Iterations / maxIter<br>\n",
    "The number of factorization runs<br>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "The models has been trained\n"
     ]
    }
   ],
   "source": [
    "from pyspark.ml.recommendation import ALS\n",
    "\n",
    "als1 = ALS(rank=3, maxIter=15,userCol=\"custId\",itemCol=\"stockCode\",ratingCol=\"purch\",implicitPrefs=True)\n",
    "model1 = als1.fit(trainDf)\n",
    "\n",
    "als2 = ALS(rank=15, maxIter=3,userCol=\"custId\",itemCol=\"stockCode\",ratingCol=\"purch\",implicitPrefs=True)\n",
    "model2 = als2.fit(trainDf)\n",
    "\n",
    "als3 = ALS(rank=15, maxIter=15,userCol=\"custId\",itemCol=\"stockCode\",ratingCol=\"purch\",implicitPrefs=True)\n",
    "model3 = als3.fit(trainDf)\n",
    "\n",
    "print \"The models has been trained\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Test the models\n",
    "\n",
    "Use the models to predict what the user will rate a certain item.  The closer our model to 1 that our model rates an item a user has already purchased, the better.\n",
    "\n",
    "#### Evaluate the model with the cross validation dataframe by using the transform function.\n",
    "\n",
    "Some of the users or purchases in the cross validation data may not have been in the training data.  Let's remove the ones that are not."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "25869\n",
      "25844\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import UserDefinedFunction\n",
    "from pyspark.sql.types import BooleanType\n",
    "customers = set(trainDf.map(lambda line: line.custId).collect())\n",
    "stock = set(trainDf.map(lambda line: line.stockCode).collect())\n",
    "\n",
    "print cvDf.count()\n",
    "cvDf = cvDf.rdd.filter(lambda line: line.stockCode in stock and\\\n",
    "                                           line.custId in customers).toDF()\n",
    "print cvDf.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(custId=14286, stockCode=20831, purch=1, prediction=0.01889532431960106), Row(custId=13949, stockCode=20831, purch=1, prediction=0.018036123365163803)]\n",
      "[Row(custId=14286, stockCode=20831, purch=1, prediction=0.04725762456655502), Row(custId=13949, stockCode=20831, purch=1, prediction=0.03975377976894379)]\n",
      "[Row(custId=14286, stockCode=20831, purch=1, prediction=0.0679328441619873), Row(custId=13949, stockCode=20831, purch=1, prediction=0.06638696044683456)]\n"
     ]
    }
   ],
   "source": [
    "predictions1 = model1.transform(cvDf)\n",
    "predictions2 = model2.transform(cvDf)\n",
    "predictions3 = model3.transform(cvDf)\n",
    "\n",
    "print predictions1.take(2)\n",
    "print predictions2.take(2)\n",
    "print predictions3.take(2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Calculate and print the Mean Squared Error.   For all ratings, subtract the prediction from the actual purchase (1), square the result, and take the mean of all of the squared differences."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean squared error = 0.7388 for our first model\n",
      "Mean squared error = 0.7003 for our second model\n",
      "Mean squared error = 0.6691 for our second model\n"
     ]
    }
   ],
   "source": [
    "meanSquaredError1 = predictions1.map(lambda line: (line.purch - line.prediction)**2).mean()\n",
    "meanSquaredError2 = predictions2.map(lambda line: (line.purch - line.prediction)**2).mean()\n",
    "meanSquaredError3 = predictions3.map(lambda line: (line.purch - line.prediction)**2).mean()\n",
    "    \n",
    "print 'Mean squared error = %.4f for our first model' % meanSquaredError1\n",
    "print 'Mean squared error = %.4f for our second model' % meanSquaredError2\n",
    "print 'Mean squared error = %.4f for our second model' % meanSquaredError3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Confirm the model by testing it with the test data and the best hyperparameters found during cross validation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Mean squared error = 0.6677 for our best model\n"
     ]
    }
   ],
   "source": [
    "filteredTestDf = testDf.rdd.filter(lambda line: line.stockCode in stock and\\\n",
    "                                              line.custId in customers).toDF()\n",
    "predictions4 = model3.transform(filteredTestDf)\n",
    "meanSquaredError4 = predictions4.map(lambda line: (line.purch - line.prediction)**2).mean()\n",
    "    \n",
    "print 'Mean squared error = %.4f for our best model' % meanSquaredError4"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Implement the model\n",
    "\n",
    "Use the best model to predict items the user will be interested in.\n",
    "\n",
    "First, create a dataframe in which each row has the user id and an item id."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(stockCode=21231, custId=15544), Row(stockCode=85231, custId=15544), Row(stockCode=22431, custId=15544), Row(stockCode=23231, custId=15544), Row(stockCode=22631, custId=15544)]\n"
     ]
    }
   ],
   "source": [
    "from pyspark.sql.functions import lit\n",
    "\n",
    "stock15544 = set(trainDf.filter(trainDf['custId'] == 15544).map(lambda line: line.stockCode).collect())\n",
    "\n",
    "userItems = trainDf.select(\"stockCode\").distinct().\\\n",
    "            withColumn('custId', lit(15544)).\\\n",
    "            rdd.filter(lambda line: line.stockCode not in stock15544).toDF()\n",
    "\n",
    "print userItems.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Use 'transform' to rate each item."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(stockCode=20831, custId=15544, prediction=0.004703751299530268), Row(stockCode=21031, custId=15544, prediction=0.010227371007204056), Row(stockCode=21231, custId=15544, prediction=0.058025069534778595), Row(stockCode=21631, custId=15544, prediction=-0.02571825124323368), Row(stockCode=22031, custId=15544, prediction=0.05723518133163452)]\n"
     ]
    }
   ],
   "source": [
    "userItems = model3.transform(userItems)\n",
    "\n",
    "print userItems.take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Print the top 5 recommendations."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(stockCode=84997, custId=15544, prediction=0.5976683497428894), Row(stockCode=21213, custId=15544, prediction=0.5788490772247314), Row(stockCode=16161, custId=15544, prediction=0.5184972286224365), Row(stockCode=22138, custId=15544, prediction=0.505846381187439), Row(stockCode=22090, custId=15544, prediction=0.49148523807525635)]\n"
     ]
    }
   ],
   "source": [
    "print userItems.sort(\"prediction\",ascending=False).take(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Let's look up this user and the recommended product ID's in the excel file...\n",
    "\n",
    "<img src='https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/user.png' width=\"80%\" height=\"80%\"></img>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "This user seems to have purchased a lot of childrens gifts and some holiday items.  The recomendation engine we created suggested some items along these lines\n",
    "\n",
    "##### The ALS algorithm uses some randomness, so the recommendations yours produces may be different than these."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "                           description\n",
      "0                   WRAP ENGLISH ROSE \n",
      "1    CHILDRENS CUTLERY POLKADOT GREEN \n",
      "2              PAPER BUNTING RETROSPOT\n",
      "3                     WRAP  PINK FLOCK\n",
      "4                    WRAP BAD HAIR DAY\n",
      "5    RED 3 PIECE RETROSPOT CUTLERY SET\n",
      "6    PINK 3 PIECE POLKADOT CUTLERY SET\n",
      "7          PACK OF 72 SKULL CAKE CASES\n",
      "8        BAKING SET 9 PIECE RETROSPOT \n",
      "9      CHILDRENS CUTLERY POLKADOT PINK\n",
      "10               WRAP SUKI AND FRIENDS\n",
      "11  GREEN 3 PIECE POLKADOT CUTLERY SET\n",
      "12     CHILDRENS CUTLERY POLKADOT BLUE\n",
      "13   BLUE 3 PIECE POLKADOT CUTLERY SET\n",
      "14    CHILDRENS CUTLERY RETROSPOT RED \n"
     ]
    }
   ],
   "source": [
    "query = \"\"\"\n",
    "SELECT \n",
    "    distinct description \n",
    "FROM \n",
    "    retailPurchases \n",
    "WHERE \n",
    "    stockCode in (84997,21213,16161,22138,22090)\n",
    "\"\"\"\n",
    "items = sqlContext.sql(query)\n",
    "print items.toPandas()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Data Citation\n",
    "Daqing Chen, Sai Liang Sain, and Kun Guo, Data mining for the online retail industry: A case study of RFM model-based customer segmentation using data mining, Journal of Database Marketing and Customer Strategy Management, Vol. 19, No. 3, pp. 197â€“208, 2012 (Published online before print: 27 August 2012. doi: 10.1057/dbm.2012.17)."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.11"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}