Lab 3 - Scalable matrix factorisation for collaborative filtering.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# COM6012 Scalable Machine Learning 2019 - Haiping Lu\n",
    "# Lab 3: Matrix factorisation for collaborative filtering recommender systems\n",
    "\n",
    "## Objectives\n",
    "\n",
    "* Task 1: To finish in the lab session. **Essential**\n",
    "* Task 2: To finish in the lab session. **Essential**\n",
    "* Task 3: To explore by yourself. **Optional but recommended**\n",
    "\n",
    "**Suggested reading**: \n",
    "* [Collaborative Filtering in Spark](https://spark.apache.org/docs/2.3.2/ml-collaborative-filtering.html)\n",
    "* [DataBricks movie recommendations tutorial](https://github.com/databricks/spark-training/blob/master/website/movie-recommendation-with-mllib.md![image.png](attachment:image.png)). [**DataBricks**](https://en.wikipedia.org/wiki/Databricks) is a company founded by the creators of Apache Spark, checking out their latest packages at [their GitHub page](https://github.com/databricks), e.g., [integration with Scikit-learn](https://github.com/databricks/spark-sklearn), [Deep Learning Pipelines for Apache Spark including TensorFlow](https://github.com/databricks/spark-deep-learning)\n",
    "* [Collaborative Filtering on Wiki](http://en.wikipedia.org/wiki/Recommender_system#Collaborative_filtering)\n",
    "* [Python API on ALS for recommender system](https://spark.apache.org/docs/2.3.2/api/python/pyspark.ml.html#pyspark.ml.recommendation.ALS)\n",
    "* Chapter 15 (particularly Section 15.3.2) of [PySpark tutorial](https://runawayhorse001.github.io/LearningApacheSpark/pyspark.pdf) \n",
    "\n",
    "[**Learn PySpark APIs via Pictures**](https://github.com/jkthompson/pyspark-pictures) (**from recommended/discover repositories** in GitHub, i.e., found via **recommender systems**!)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://github.com/haipinglu/ScalableML/"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Finish Lab 1 and Lab 2 sessions\n",
    "\n",
    "### You will need to make use of lab 1 and lab 2 notebooks for the Quiz in Lab 4.\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Movie recommendation via collaborative filtering\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Basic setup unless using shell"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "#import findspark\n",
    "#findspark.init()\n",
    "import pyspark\n",
    "\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder \\\n",
    "    .master(\"local[2]\") \\\n",
    "    .appName(\"COM6012 Collaborative Filtering RecSys\") \\\n",
    "    .getOrCreate()\n",
    "\n",
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Collaborative filtering\n",
    "[Collaborative filtering](http://en.wikipedia.org/wiki/Recommender_system#Collaborative_filtering) is a classic approach for recommender systems. These techniques aim to fill in the missing entries of a user-item association matrix primarily based on the matrix *itself*.  `spark.ml` currently supports **model-based** collaborative filtering, in which users and products are described by a small set of latent factors that can be used to predict missing entries, using the **alternating least squares (ALS)** algorithm. \n",
    "\n",
    "API: `class pyspark.ml.recommendation.ALS(rank=10, maxIter=10, regParam=0.1, numUserBlocks=10, numItemBlocks=10, implicitPrefs=False, alpha=1.0, userCol='user', itemCol='item', seed=None, ratingCol='rating', nonnegative=False, checkpointInterval=10, intermediateStorageLevel='MEMORY_AND_DISK', finalStorageLevel='MEMORY_AND_DISK', coldStartStrategy='nan')`\n",
    "\n",
    "The following parameters are available:\n",
    "- *rank*: the number of latent factors in the model (defaults to 10).\n",
    "- *maxIter* is the maximum number of iterations to run (defaults to 10).\n",
    "- *regParam*: the regularization parameter in ALS (defaults to 1.0).\n",
    "- *numUserBlocks*/*numItemBlocks*: the number of blocks the users and items will be partitioned into in order to parallelize computation (defaults to 10).\n",
    "- *implicitPrefs*: whether to use the explicit feedback ALS variant or one adapted for implicit feedback data (defaults to false which means using explicit feedback).\n",
    "- *alpha*: a parameter applicable to the implicit feedback variant of ALS that governs the baseline confidence in preference observations (defaults to 1.0).\n",
    "- *nonnegative*: whether or not to use nonnegative constraints for least squares (defaults to false).\n",
    "- *coldStartStrategy*: can be set to “drop” in order to drop any rows in the DataFrame of predictions that contain NaN values (defaults to \"nan\", assigning NaN to a user and/or item factor is not present in the model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Movie recommendation\n",
    "\n",
    "In the cells below, we present a small example of collaborative filtering with the data taken from the [MovieLens](http://grouplens.org/datasets/movielens/) project. In this notebook, we use the old 100k dataset (already downloaded in the `Data` folder but you are encouraged to view the source.\n",
    "\n",
    "The dataset looks like this:\n",
    "\n",
    "    196     242     3       881250949\n",
    "    186     302     3       891717742\n",
    "    22      377     1       878887116\n",
    "    244     51      2       880606923\n",
    "    ...\n",
    "\n",
    "This is a **tab separated** list of \n",
    "    \n",
    "    user id | item id | rating | timestamp "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "####  Explicit vs. implicit feedback\n",
    "\n",
    "The data above is typically viewed as a user-item matrix with the ratings as the entries and users and items determine the row and column indices. The ratings are **explicit feedback**. The *Mean Squared Error* of rating prediction can be used to evaluate the recommendation model.\n",
    "\n",
    "The ratings can also be used differently. We can treat them as  treated as numbers representing the strength in observations of user actions, i.e., as **implicit feedback** similar to the number of clicks, or the cumulative duration someone spent viewing a movie. Such numbers are then related to the level of confidence in observed user preferences, rather than explicit ratings given to items. The model then tries to find latent factors that can be used to predict the expected preference of a user for an item."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Cold-start problem\n",
    "\n",
    "The cold-start problem refers to the cases when some users and/or items in the test dataset were not present during training the model. In Spark, these users and items are either assigned `NaN` (not a number, default) or dropped (option `drop`).\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.ml.evaluation import RegressionEvaluator\n",
    "from pyspark.ml.recommendation import ALS\n",
    "from pyspark.sql import Row"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Read the data in and split words (tab separated)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "lines = spark.read.text(\"Data/MovieLens100k.data\").rdd\n",
    "parts = lines.map(lambda row: row.value.split(\"\\t\"))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We need to convert the text (`String`) into numbers (`int` or `float`) and then convert RDD to DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/haiping/anaconda3/lib/python3.6/importlib/_bootstrap.py:219: RuntimeWarning: numpy.dtype size changed, may indicate binary incompatibility. Expected 96, got 88\n",
      "  return f(*args, **kwds)\n"
     ]
    }
   ],
   "source": [
    "ratingsRDD = parts.map(lambda p: Row(userId=int(p[0]), movieId=int(p[1]),rating=float(p[2]), timestamp=int(p[3])))\n",
    "ratings = spark.createDataFrame(ratingsRDD)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[The warning is benign](https://stackoverflow.com/questions/40845304/runtimewarning-numpy-dtype-size-changed-may-indicate-binary-incompatibility)\n",
    "\n",
    "Check data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------+---------+------+\n",
      "|movieId|rating|timestamp|userId|\n",
      "+-------+------+---------+------+\n",
      "|    242|   3.0|881250949|   196|\n",
      "|    302|   3.0|891717742|   186|\n",
      "|    377|   1.0|878887116|    22|\n",
      "|     51|   2.0|880606923|   244|\n",
      "|    346|   1.0|886397596|   166|\n",
      "+-------+------+---------+------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ratings.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Check data type"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- movieId: long (nullable = true)\n",
      " |-- rating: double (nullable = true)\n",
      " |-- timestamp: long (nullable = true)\n",
      " |-- userId: long (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "ratings.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Prepare the training/test data."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "(training, test) = ratings.randomSplit([0.8, 0.2])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Build the recommendation model using ALS on the training data. Note we set cold start strategy to `drop` to ensure we don't get NaN evaluation metrics."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "als = ALS(maxIter=10, regParam=0.1, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\",\n",
    "          coldStartStrategy=\"drop\")\n",
    "model = als.fit(training)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Evaluate the model by computing the RMSE on the test data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root-mean-square error = 0.9252028388998063\n"
     ]
    }
   ],
   "source": [
    "predictions = model.transform(test)\n",
    "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\",predictionCol=\"prediction\")\n",
    "rmse = evaluator.evaluate(predictions)\n",
    "print(\"Root-mean-square error = \" + str(rmse))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Generate top 10 movie recommendations for each user"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "userRecs = model.recommendForAllUsers(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|userId|recommendations                                                                                                                                                                      |\n",
      "+------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|471   |[[1260, 5.283116], [1469, 5.1202607], [1166, 4.7839737], [1427, 4.674018], [916, 4.6624784], [1410, 4.658369], [1167, 4.569021], [309, 4.4751625], [465, 4.3423076], [862, 4.339869]]|\n",
      "|463   |[[694, 4.4602747], [19, 4.276263], [116, 4.2269835], [1344, 4.2179418], [113, 4.1964374], [1137, 4.161138], [889, 4.127839], [963, 4.118122], [1167, 4.110317], [1278, 4.095075]]    |\n",
      "|833   |[[1536, 4.4497566], [1597, 4.419347], [512, 4.3735576], [1368, 4.3057766], [179, 4.3039703], [1643, 4.281577], [896, 4.2766156], [192, 4.26899], [32, 4.2284794], [180, 4.223832]]   |\n",
      "|496   |[[913, 4.6929584], [1368, 4.619412], [1240, 4.5141835], [390, 4.475494], [1589, 4.420513], [838, 4.323143], [1367, 4.316384], [1005, 4.297203], [1137, 4.290421], [253, 4.269342]]   |\n",
      "|148   |[[344, 5.31927], [1463, 5.290077], [408, 5.2523417], [613, 5.1700068], [1062, 5.167716], [306, 5.1382937], [958, 5.1373463], [838, 5.070853], [361, 5.0225253], [114, 4.988438]]     |\n",
      "+------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "userRecs.show(5,  False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Generate top 10 user recommendations for each movie"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|movieId|recommendations                                                                                                                                                                     |\n",
      "+-------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|1580   |[[36, 1.1969261], [519, 1.0878557], [340, 1.044994], [55, 1.0379616], [532, 0.99146795], [134, 0.9861098], [507, 0.9831521], [137, 0.97840774], [152, 0.97709715], [575, 0.9743821]]|\n",
      "|471    |[[688, 5.151984], [811, 4.982617], [507, 4.891306], [849, 4.808163], [628, 4.8041553], [636, 4.686763], [939, 4.679654], [4, 4.606876], [907, 4.5915523], [372, 4.59049]]           |\n",
      "|1591   |[[174, 4.7854176], [34, 4.7753053], [427, 4.743956], [519, 4.743409], [628, 4.651128], [440, 4.638868], [341, 4.6130295], [861, 4.4738364], [928, 4.457759], [242, 4.4504]]         |\n",
      "|463    |[[697, 4.864459], [46, 4.7283297], [173, 4.6437197], [148, 4.624897], [4, 4.603298], [849, 4.5984817], [523, 4.573879], [770, 4.514226], [808, 4.507024], [939, 4.4809346]]         |\n",
      "|833    |[[507, 4.5314403], [636, 4.395785], [324, 4.3616643], [887, 4.31266], [810, 4.3015733], [534, 4.2368813], [688, 4.1852], [546, 4.1067705], [783, 4.1019306], [261, 4.0984893]]      |\n",
      "+-------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "only showing top 5 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movieRecs = model.recommendForAllItems(10)\n",
    "movieRecs.show(5, False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Generate top 10 movie recommendations for a specified set of users"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------+\n",
      "|userId|\n",
      "+------+\n",
      "|    26|\n",
      "|    29|\n",
      "|   474|\n",
      "+------+\n",
      "\n",
      "+------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|userId|recommendations                                                                                                                                                                      |\n",
      "+------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|26    |[[1643, 3.9734263], [1449, 3.9256008], [1122, 3.8582377], [1398, 3.838931], [119, 3.838321], [64, 3.8263464], [1064, 3.818812], [1450, 3.8132286], [963, 3.809756], [1642, 3.741221]]|\n",
      "|474   |[[1643, 5.4205756], [1449, 5.056973], [1463, 4.978273], [119, 4.888391], [64, 4.8586493], [318, 4.8300595], [1122, 4.8250628], [1398, 4.7978935], [408, 4.749604], [483, 4.7439947]] |\n",
      "|29    |[[1398, 4.7427278], [119, 4.6581464], [1449, 4.5915427], [19, 4.579688], [361, 4.5551777], [1642, 4.552145], [272, 4.528605], [958, 4.5273533], [60, 4.518589], [1405, 4.5175357]]   |\n",
      "+------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "users = ratings.select(als.getUserCol()).distinct().limit(3)\n",
    "userSubsetRecs = model.recommendForUserSubset(users, 10)\n",
    "users.show()\n",
    "userSubsetRecs.show(3,False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Generate top 10 user recommendations for a specified set of movies"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-------+\n",
      "|movieId|\n",
      "+-------+\n",
      "|    474|\n",
      "|     29|\n",
      "|     26|\n",
      "+-------+\n",
      "\n",
      "+-------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|movieId|recommendations                                                                                                                                                                |\n",
      "+-------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "|26     |[[849, 4.3745384], [252, 4.3024592], [270, 4.2542305], [152, 4.241466], [38, 4.2352576], [477, 4.234951], [808, 4.2113433], [907, 4.20929], [557, 4.2038007], [462, 4.1931267]]|\n",
      "|474    |[[118, 5.1251917], [219, 5.044167], [556, 5.042454], [274, 5.0016336], [794, 4.994564], [808, 4.976072], [592, 4.9716034], [583, 4.953826], [686, 4.92491], [239, 4.9227204]]  |\n",
      "|29     |[[507, 4.6093554], [887, 4.3661084], [810, 4.36563], [811, 4.247795], [636, 4.2450175], [324, 4.2134075], [849, 4.136991], [534, 4.13638], [620, 4.128891], [939, 4.042479]]   |\n",
      "+-------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+\n",
      "\n"
     ]
    }
   ],
   "source": [
    "movies = ratings.select(als.getItemCol()).distinct().limit(3)\n",
    "movieSubSetRecs = model.recommendForItemSubset(movies, 10)\n",
    "movies.show()\n",
    "movieSubSetRecs.show(3,False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Exercise - Further analysis of the MovieLens data (completing two or more questions is considered as completion of this exercise).\n",
    "* Consider more parameter settings to observe the effecttsm e.g., different values of *rank* and/or *regParam*, `nan` vs `drop` for `coldStartStrategy`, etc.\n",
    "* Use cross validation to select the best model among various parameter settings (reference: Section 15.3.2 of [PySpark tutorial](https://runawayhorse001.github.io/LearningApacheSpark/pyspark.pdf) )\n",
    "* Create a standalone program that carries out collaborative filtering. Run this on a bigger [MovieLens dataset](http://grouplens.org/datasets/movielens/), e.g., 1M, 10M or 20M.\n",
    "\n",
    "* Use 10-fold cross validation (with an 80% training and 20% testing split) to find an average mean average (or squared) error on your test data. Keep your program as parallel as possible. You can create your splits randomly (or any other way you choose!), and don't forget who has access to various variables and who doesn't..."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. More Recommender Systems via ALS (Optional but recommended)\n",
    "\n",
    "### Databricks tutorial\n",
    "* Complete the tasks in [quiz provided by DataBricks](https://github.com/databricks/spark-training/blob/master/machine-learning/python/MovieLensALS.py) on their data or the data from MovieLens directly. [Solution](https://github.com/databricks/spark-training/blob/master/machine-learning/python/solution/MovieLensALS.py) is posted but you are suggested to try before consulting the solution.\n",
    "\n",
    "### Santander Kaggle competition on produce recommendation\n",
    "* A recent Kaggle competition on [Santander Product Recommendation](https://www.kaggle.com/c/santander-product-recommendation) with a prize of **USD 60,000**, and **1,787 teams** participating. \n",
    "* Follow this [PySpark notebook on an ALS-based solution](https://www.elenacuoco.com/2016/12/22/alternating-least-squares-als-spark-ml/)\n",
    "* Learn the way to consider **implcit preferences** and do the same for other recommendation problems.\n",
    "\n",
    "\n",
    "### Stock Portfolio Recommendations\n",
    "* Follow Chapter 15 of [PySpark tutorial](https://runawayhorse001.github.io/LearningApacheSpark/pyspark.pdf)  to perform [Stock Portfolio Recommendations](https://en.wikipedia.org/wiki/Portfolio_investment))\n",
    "* The data can be downloaded from [Online Retail Data Set](https://archive.ics.uci.edu/ml/datasets/online+retail) at UCI. \n",
    "* Please pay attention to the **data cleaning** step that removes rows containing null value. You may need to do the same when you are dealing with real data.\n",
    "* The data manipulation steps are useful to learn.\n",
    "\n",
    "### Context-aware recommendation\n",
    "\n",
    "* See the method in [Joint interaction with context operation for collaborative filtering](https://www.sciencedirect.com/science/article/pii/S0031320318304242?dgcid=rss_sd_all) and implement it in PySpark\n",
    "* Perform the **time split recommendation** as disscussed in the paper for the above recommender systems.\n"
   ]
  }
 ],
 "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
}