Bonus - Intro to Spark Machine Learning.ipynb

{
    "nbformat_minor": 1, 
    "metadata": {
        "kernelspec": {
            "name": "python2-spark20", 
            "display_name": "Python 2 with Spark 2.0", 
            "language": "python"
        }, 
        "language_info": {
            "mimetype": "text/x-python", 
            "version": "2.7.11", 
            "name": "python", 
            "pygments_lexer": "ipython2", 
            "nbconvert_exporter": "python", 
            "file_extension": ".py", 
            "codemirror_mode": {
                "name": "ipython", 
                "version": 2
            }
        }
    }, 
    "cells": [
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "![](https://github.com/krondor/data-science-pot/blob/master/dsx_logo.png?raw=true)\n\n# Introduction to Apache Spark: Machine Learning\n\nIn this notebook, you'll learn how to create a model for purchase recommendations using the alternating least squares algorithm of the Apache Spark machine learning library. Machine learning is based on algorithms that can learn from data without relying on rules-based programming.  \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\nThis notebook uses pySpark, the Python API for Spark. Some knowledge of Python is recommended. This notebook runs on Python 2 with Spark 1.6 and Spark 2.0.\n\nIf you are new to Apache Spark, see the first two parts of this lab: \n - Introduction to Apache Spark lab, part 1: Basic concepts. (need link when published)\n - Introduction to Apache Spark lab, part 2: Querying data. (need link when published) \n\n## Spark machine learning library\n[Apache Spark\u2019s machine learning library](http://spark.apache.org/docs/latest/mllib-guide.html) makes practical machine learning scalable and easy. The library consists of common machine learning algorithms and utilities, including classification, regression, clustering, collaborative filtering (this notebook!), dimensionality reduction, lower-level optimization primitives, and higher-level pipeline APIs.\n\nThe library has two packages:\n- spark.mllib contains the original API that handles data in RDDs. It's in maintenance mode, but fully supported.\n- spark.ml contains a newer API for constructing ML pipelines. It handles data in DataFrames. It's being actively enhanced.\n\n## Alternating least squares algorithm\nThe alternating least squares (ALS) algorithm provides collaborative filtering between customers and products to find products that the customers might like, based on their previous purchases or ratings.\n\nThe ALS algorithm creates a matrix of all customers versus all products. Most cells in the matrix are empty, which means the customer hasn't bought that product. The ALS algorithm then fills in the probability of customers buying products that they haven't bought yet, based on the similarities between customer purchases and the similarities between products. The algorithm uses the least squares computation to minimize the estimation errors and alternates between fixing the customer factors and solving for product factors and fixing the product factors and solving for customer factors.\n\nYou don't, however, need to understand how the ALS algorithm works to use it! Spark machine learning algorithms have default values that work well in most cases."
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "## Table of contents\n\n1. [Get the data](#getdata)<br>\n2. [Prepare and shape the data](#prepare)<br>\n    2.1 [Format the data](#prepare1)<br>\n    2.2 [Clean the data](#prepare2)<br>\n    2.3 [Create a DataFrame](#prepare3)<br>\n    2.4 [Remove unneeded columns](#prepare4)<br>\n3. [Split the data into three sets](#split)<br>\n4. [Build recommendation models](#model)<br>\n5. [Test the models](#test)<br>\n    5.1 [Clean the cross validation data set](#test1)<br>\n    5.2 [Run the models on the cross validation data set](#test2)<br>\n    5.3 [Calculate the accuracy for each model](#test3)<br>\n    5.4 [Confirm the best model](#test4)<br>\n6. [Implement the mode](#implement)<br>\n    6.1 [Create a DataFrame for the customer and all products](#implement1)<br>\n    6.2 [Rate each product](#implement2)<br>\n    6.3 [Find the top recommendations](#implement3)<br>\n    6.4 [Compare purchased and recommended products](#implement4)<br>\n7. [Summary and next steps](#summary)"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"getdata\"></a>\n## 1. Get the data \nThe data set contains the transactions of an online retailer of gift items for 01/12/2010 - 09/12/2011. Many of the customers are wholesalers.\n\nYou'll be using a slightly modified version of this data set: [http://archive.ics.uci.edu/ml/datasets/Online+Retail](http://archive.ics.uci.edu/ml/datasets/Online+Retail).  \n\nHere's a glimpse of the data:\n\n<img src='https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/FullFile.png' width=\"80%\" height=\"80%\"></img>"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "Download the CSV version of the data set, from which the commas in the product descriptions are removed:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "!rm 'OnlineRetail.csv.gz' -f\n!wget https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/OnlineRetail.csv.gz"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "Put the data into an RDD and print the first 5 rows:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "loadRetailData = sc.textFile(\"OnlineRetail.csv.gz\")\n\nfor row in loadRetailData.take(5):\n    print row"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "Each row in the RDD is a string that correlates to a line in the CSV file."
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"prepare\"></a>\n## 2. Prepare and shape the data\n\nIt's been said that preparing and shaping data is 80% of a data scientist's job. Having the right data in the right format is critical for getting accurate results.\n\nTo get the data ready, complete these tasks:\n\n1. [Format the data](#prepare1)\n1. [Clean the data](#prepare2)\n1. [Create a DataFrame](#prepare3)\n1. [Remove unneeded columns](#prepare4)"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"prepare1\"></a>\n### 2.1 Format the data\nRemove the header from the RDD and split the string in each row with a comma:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "header = loadRetailData.first()\nloadRetailData = loadRetailData.filter(lambda line: line != header).\\\n                            map(lambda l: l.split(\",\"))\n\nfor row in loadRetailData.take(5):\n    print row"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"prepare2\"></a>\n### 2.2 Clean the data\nRemove the rows that have incomplete data. Keep only the rows that meet the following criteria:\n - The purchase quantity is greater than 0 \n - The customerID not equal to 0 \n - The stock code is not blank after you remove non-numeric characters"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "import re\n\nloadRetailData = loadRetailData.filter(lambda l: int(l[3]) > 0\\\n                                and len(re.sub(\"\\D\", \"\", l[1])) != 0 \\\n                                and len(l[6]) != 0)\n\nprint loadRetailData.take(2)"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"prepare3\"></a>\n### 2.3 Create a DataFrame\n\nFirst, create an SQLContext and map each line to a row: "
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "from pyspark.sql import SQLContext, Row\nsqlContext = SQLContext(sc)\n\n#Convert each line to a Row.\nloadRetailData = 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]))"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "Create a DataFrame and show the inferred schema:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "retailDf = sqlContext.createDataFrame(loadRetailData)\nprint retailDf.printSchema()"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "Register the DataFrame as a table so that you can run SQL queries on it and show the first two rows:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "retailDf.registerTempTable(\"retailPurchases\")\nsqlContext.sql(\"SELECT * FROM retailPurchases limit 2\").toPandas()"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"prepare4\"></a>\n### 2.4 Remove unneeded columns\nThe only columns you need are `custId`, `stockCode`, and a new column, `purch`, which has a value of 1 to indicate that the customer purchased the product:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "query = \"\"\"\nSELECT \n    custId, stockCode, 1 as purch\nFROM \n    retailPurchases \ngroup \n    by custId, stockCode\"\"\"\nretailDf = sqlContext.sql(query)\nretailDf.registerTempTable(\"retailDf\")\n\nsqlContext.sql(\"select * from retailDf limit 10\").toPandas()"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"split\"></a>\n## 3. Split the data into three sets\nYou'll split the data into three sets: \n - a testing data set (10% of the data)\n - a cross validation data set (10% of the data)\n - a training data set (80% of the data)\n\nSplit the data randomly and create a DataFrame for each data set:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "testDf, cvDf, trainDf = retailDf.randomSplit([.1,.1,.8],1)\n\nprint \"trainDf count: \", trainDf.count(), \" example: \"\nfor row in trainDf.take(2): print row\nprint\n\nprint \"cvDf count: \", cvDf.count(), \" example: \"\nfor row in cvDf.take(2): print row\nprint\n\nprint \"testDf count: \", testDf.count(), \" example: \"\nfor row in testDf.take(2): print row\nprint"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"model\"></a>\n## 4. Build recommendation models\nMachine learning algorithms have standard parameters and hyperparameters. Standard parameters specify data and options. Hyperparameters control the performance of the algorithm.\n\nThe ALS algorithm has these hyperparameters:  \n\n - The `rank` hyperparameter represents the number of features. The default value of `rank` is 10.\n - The `maxIter` hyperparameter represents the number of iterations to run the least squares computation. The default value of `maxIter` is 10.\n\nUse the training DataFrame to train three models with the ALS algorithm with different values for the `rank` and `maxIter` hyperparameters. Assign the `userCol`, `itemCol`, and `ratingCol` parameters to the appropriate data columns. Set the `implicitPrefs` parameter to `true` so that the algorithm can predict latent factors."
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {
                "scrolled": true
            }, 
            "cell_type": "code", 
            "source": "from pyspark.ml.recommendation import ALS\n\nals1 = ALS(rank=3, maxIter=15,userCol=\"custId\",itemCol=\"stockCode\",ratingCol=\"purch\",implicitPrefs=True)\nmodel1 = als1.fit(trainDf)\n\nals2 = ALS(rank=15, maxIter=3,userCol=\"custId\",itemCol=\"stockCode\",ratingCol=\"purch\",implicitPrefs=True)\nmodel2 = als2.fit(trainDf)\n\nals3 = ALS(rank=15, maxIter=15,userCol=\"custId\",itemCol=\"stockCode\",ratingCol=\"purch\",implicitPrefs=True)\nmodel3 = als3.fit(trainDf)\n\nprint \"The models are trained\""
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"test\"></a>\n## 5. Test the models\n\nFirst test the three models on the cross validation data set and then on the testing data set. \n\nYou'll know the model is accurate when the prediction values for products that the customers have already bought is close to 1. \n\n<a id=\"test1\"></a>\n### 5.1 Clean the cross validation data set\n\nRemove any of the customers or products in the cross validation set that are not in the training set:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "from pyspark.sql.functions import UserDefinedFunction\nfrom pyspark.sql.types import BooleanType\ncustomers = set(trainDf.rdd.map(lambda line: line.custId).collect())\nstock = set(trainDf.rdd.map(lambda line: line.stockCode).collect())\n\nprint cvDf.count()\ncvDf = cvDf.rdd.filter(lambda line: line.stockCode in stock and\\\n                                           line.custId in customers).toDF()\nprint cvDf.count()"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"test2\"></a>\n### 5.2 Run the models on the cross validation data set\nRun the model with the cross validation DataFrame by using the `transform` function and print the first two rows of each set of predictions:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {
                "scrolled": false
            }, 
            "cell_type": "code", 
            "source": "predictions1 = model1.transform(cvDf)\npredictions2 = model2.transform(cvDf)\npredictions3 = model3.transform(cvDf)\n\nprint predictions1.take(2)\nprint predictions2.take(2)\nprint predictions3.take(2)"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"test3\"></a>\n### 5.3 Calculate the accuracy for each model  \n\nYou'll use the mean squared error calculation to determine accuracy by comparing the prediction values for products to the actual purchase values. Remember that if a customer purchased a product, the value in the `purch` column is 1. The mean squared error calculation measures the average of the squares of the errors between what is estimated and the existing data. The lower the mean squared error value, the more accurate the model. \n\nFor all predictions, subtract the prediction from the actual purchase value (1), square the result, and calculate the mean of all of the squared differences:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "meanSquaredError1 = predictions1.rdd.map(lambda line: (line.purch - line.prediction)**2).mean()\nmeanSquaredError2 = predictions2.rdd.map(lambda line: (line.purch - line.prediction)**2).mean()\nmeanSquaredError3 = predictions3.rdd.map(lambda line: (line.purch - line.prediction)**2).mean()\n    \nprint 'Mean squared error = %.4f for our first model' % meanSquaredError1\nprint 'Mean squared error = %.4f for our second model' % meanSquaredError2\nprint 'Mean squared error = %.4f for our third model' % meanSquaredError3"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "The third model (model3) has the lowest mean squared error value, so it's the most accurate.\n\nNotice that of the three models, model3 has the highest values for the hyperparameters. At this point you might be tempted to run the model with even higher values for `rank` and `maxIter`. However, you might not get better results. Increasing the values of the hyperparameters increases the time for the model to run. Also, you don't want to overfit the model so that it exactly fits the original data. In that case, you don't get any recommendations! For best results, keep the values of the hyperparameters close to the defaults.\n\n<a id=\"test4\"></a>\n### 5.4 Confirm the best model \n\nNow run model3 on the testing data set to confirm that it's the best model. You want to make sure that the model is not over-matched to the cross validation data. It's possible for a model to match one subset of the data well but not another. If the values of the mean squared error for the testing data set and the cross validation data set are close, then you've confirmed that the model works for all the data.\n\nClean the testing data set, run model3 on the testing data set, and calculate the mean squared error:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "filteredTestDf = testDf.rdd.filter(lambda line: line.stockCode in stock and\\\n                                              line.custId in customers).toDF()\npredictions4 = model3.transform(filteredTestDf)\nmeanSquaredError4 = predictions4.rdd.map(lambda line: (line.purch - line.prediction)**2).mean()\n    \nprint 'Mean squared error = %.4f for our best model' % meanSquaredError4"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "That's pretty close. The model works for all the data."
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"implement\"></a>\n## 6. Implement the model\n\nUse the best model to predict which products that a particular customer might be interested in.\n\n<a id=\"implement1\"></a>\n### 6.1 Create a DataFrame for the customer and all products \n\nCreate a DataFrame in which each row has the customer ID (15544) and a product ID:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "from pyspark.sql.functions import lit\n\nstock15544 = set(trainDf.filter(trainDf['custId'] == 15544).rdd.map(lambda line: line.stockCode).collect())\n\nuserItems = trainDf.select(\"stockCode\").distinct().\\\n            withColumn('custId', lit(15544)).\\\n            rdd.filter(lambda line: line.stockCode not in stock15544).toDF()\n\nfor row in userItems.take(5):\n    print row.stockCode, row.custId"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"implement2\"></a>\n### 6.2 Rate each product\n\nRun the `transform` function to create a prediction value for each product:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "userItems = model3.transform(userItems)\n\nfor row in userItems.take(5):\n    print row.stockCode, row.custId, row.prediction"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"implement3\"></a>\n### 6.3 Find the top recommendations\n\nPrint the top 5 product recommendations:"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": "userItems.registerTempTable(\"predictions\")\nquery = \"select * from predictions order by prediction desc limit 5\"\n\nsqlContext.sql(query).toPandas()"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"implement4\"></a>\n### 6.4 Compare purchased and recommended products\n\nHere's a view of the products that customer 15544 bought:\n\n<img src='https://raw.githubusercontent.com/rosswlewis/RecommendationPoT/master/user.png' width=\"80%\" height=\"80%\"></img>"
        }, 
        {
            "metadata": {
                "collapsed": true
            }, 
            "cell_type": "markdown", 
            "source": "This customer bought lots of childrens gifts and some holiday items. \n\nLook at the descriptions of the recommended products to see if they are in the same categories.\n\n<div class=\"alert alert-block alert-info\">Note: The ALS algorithm uses some randomness, so the recommendations you got might be different than these.</div>"
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {
                "scrolled": true
            }, 
            "cell_type": "code", 
            "source": "stockItems = sqlContext.sql(\"select distinct stockCode, description from retailPurchases\")\nstockItems.registerTempTable(\"stockItems\")\n\nquery = \"\"\"\nselect \n    predictions.*,\n    stockItems.description\nfrom\n    predictions\ninner join stockItems on\n    predictions.stockCode = stockItems.stockCode\norder by predictions.prediction desc\nlimit 10\n\"\"\"\nsqlContext.sql(query).toPandas()"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "The recommended products look pretty similar to the purchased products, and, in some cases, are actually the same. Your model works!"
        }, 
        {
            "metadata": {}, 
            "cell_type": "markdown", 
            "source": "<a id=\"summary\"></a>\n## 7. Summary and next steps\nYou created a predictive model that makes product recommendations for customers and verified that it works.\n\nDig deeper:\n - [Collaborative Filtering](http://spark.apache.org/docs/latest/mllib-collaborative-filtering.html)\n - [Spark Machine Learning Library (MLlib) Guide](http://spark.apache.org/docs/latest/ml-guide.html)\n - [Spark Python API Docs](http://spark.apache.org/docs/latest/api/python/index.html)\n\n\n## Author\n\n\n## Data citation\nDaqing 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\u00e2\u20ac\u201c208, 2012 (Published online before print: 27 August 2012. doi: 10.1057/dbm.2012.17)."
        }, 
        {
            "outputs": [], 
            "execution_count": null, 
            "metadata": {}, 
            "cell_type": "code", 
            "source": ""
        }
    ], 
    "nbformat": 4
}