notebooks/MF_Recommender_Systems.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "name": "Lecture_13_Recommender_Systems.ipynb",
      "provenance": [],
      "collapsed_sections": [],
      "machine_shape": "hm"
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "6r8rVH_wrVFN",
        "colab_type": "text"
      },
      "source": [
        "# **Global Constants**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "_GumGWC6reG6",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "JAVA_HOME = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "GDRIVE_DIR = \"/content/gdrive\"\n",
        "GDRIVE_HOME_DIR = GDRIVE_DIR + \"/My Drive\"\n",
        "GDRIVE_DATA_DIR = GDRIVE_HOME_DIR + \"/Teaching/2019-20-BDC/datasets\"\n",
        "DATASET_URL = \"https://github.com/gtolomei/big-data-computing/raw/master/datasets/movielens-ratings-25m.csv.bz2\"\n",
        "GDRIVE_DATASET_FILE = GDRIVE_DATA_DIR + \"/\" + DATASET_URL.split(\"/\")[-1]\n",
        "\n",
        "RANDOM_SEED = 42 # for reproducibility"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "iNQWYxnsMqQv",
        "colab_type": "text"
      },
      "source": [
        "# **Spark + Google Colab Setup**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "tFudfLCRNvXT",
        "colab_type": "text"
      },
      "source": [
        "## **1.** Install PySpark and related dependencies"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "l6vtwpPGKn9W",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!pip install pyspark\n",
        "!pip install -U -q PyDrive\n",
        "!apt install openjdk-8-jdk-headless -qq\n",
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = JAVA_HOME"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YOrmY8FiOMwa",
        "colab_type": "text"
      },
      "source": [
        "## **2.** Import useful Python packages"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Fh8zPg5APmYv",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "import requests\n",
        "import pandas as pd\n",
        "import numpy as np\n",
        "import matplotlib.pyplot as plt\n",
        "import seaborn as sns\n",
        "%matplotlib inline\n",
        "\n",
        "import pyspark\n",
        "from pyspark.sql import *\n",
        "from pyspark.sql.types import *\n",
        "from pyspark.sql.functions import *\n",
        "from pyspark import SparkContext, SparkConf"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hbT2rM-4tvnB",
        "colab_type": "text"
      },
      "source": [
        "## **3.** Create Spark context"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xXQOJijlEz3I",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# create the session\n",
        "conf = SparkConf().set(\"spark.ui.port\", \"4050\").set('spark.executor.memory', '4G').set('spark.driver.memory', '45G').set('spark.driver.maxResultSize', '10G')\n",
        "\n",
        "# create the context\n",
        "sc = pyspark.SparkContext(conf=conf)\n",
        "spark = SparkSession.builder.getOrCreate()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "HV1RghvXuwQG",
        "colab_type": "text"
      },
      "source": [
        "## **4.** Create <code>ngrok</code> tunnel to check the Spark UI"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "BB4QVvEQuIXW",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "!wget https://bin.equinox.io/c/4VmDzA7iaHb/ngrok-stable-linux-amd64.zip\n",
        "!unzip ngrok-stable-linux-amd64.zip\n",
        "get_ipython().system_raw('./ngrok http 4050 &')\n",
        "!curl -s http://localhost:4040/api/tunnels | python3 -c \\\n",
        "    \"import sys, json; print(json.load(sys.stdin)['tunnels'][0]['public_url'])\""
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "on0pzpMiOA_L",
        "colab_type": "text"
      },
      "source": [
        "## **5.** Link Colab to our Google Drive"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "yzUd3JdGS1Tp",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Point Colaboratory to our Google Drive\n",
        "\n",
        "from google.colab import drive\n",
        "\n",
        "drive.mount(GDRIVE_DIR, force_remount=True)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "_eRvyl2xwCV6",
        "colab_type": "text"
      },
      "source": [
        "## **6.** Check everything is ok"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "4fqJ5f0JE3BL",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "spark"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "qhTN342EEOYZ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "sc._conf.getAll()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Mu77QbD2vC_o",
        "colab_type": "text"
      },
      "source": [
        "# **MovieLens Recommender System**\n",
        "\n",
        "In this notebook, we will be using a dataset from [GroupLens](https://grouplens.org/datasets/movielens/), which collected and made available rating data sets from the [MovieLens web site](http://movielens.org). \n",
        "\n",
        "More specifically, this dataset - released on December 2019 - is the **MovieLens 25M Dataset** containing **25 million** ratings applied to $62,000$ movies by $162,000$ users (using 5-star rating system). \n",
        "In addition to movie ratings, the original collection includes also **1 million** tags and tag genome data with **15 million** relevance scores across $1,129$ tags, which however we will not be using here. Anyway, in case you are interested in working also with those data, the whole collection can be downloaded from this [link](http://files.grouplens.org/datasets/movielens/ml-25m.zip).\n",
        "\n",
        "The task is to provide movie recommendations to users that they are likely to be interested in and engage with."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "g-vLczwIuUhG",
        "colab_type": "text"
      },
      "source": [
        "# **1. Data Collection**\n",
        "\n",
        "This is the first step we need to accomplish before going any further. The dataset will be downloaded directly to our Google Drive, as usual."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "j5fI3wiGvKBl",
        "colab_type": "text"
      },
      "source": [
        "### **Download dataset file from URL directly to our Google Drive**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "kxrLDE_4e7KH",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def get_data(dataset_url, dest, chunk_size=1024):\n",
        "  response = requests.get(dataset_url, stream=True)\n",
        "  if response.status_code == 200:\n",
        "    with open(dest, \"wb\") as file:\n",
        "      for block in response.iter_content(chunk_size=chunk_size): \n",
        "        if block: \n",
        "          file.write(block)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "quiuGbfyv8vT",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "print(\"Retrieving dataset from URL: {} ...\".format(DATASET_URL))\n",
        "get_data(DATASET_URL, GDRIVE_DATASET_FILE)\n",
        "print(\"Dataset successfully retrieved and stored at: {}\".format(GDRIVE_DATASET_FILE))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DevlrMcPw1ZI",
        "colab_type": "text"
      },
      "source": [
        "### **Read dataset file into a Spark Dataframe**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "qKi5Hd60FFcX",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "ratings_df = spark.read.load(GDRIVE_DATASET_FILE, \n",
        "                         format=\"csv\", \n",
        "                         sep=\",\", \n",
        "                         inferSchema=\"true\", \n",
        "                         header=\"true\"\n",
        "                         )"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "IPTd8ep9x74H",
        "colab_type": "text"
      },
      "source": [
        "### **Check the shape of the loaded dataset, i.e., number of rows and columns**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "JyRyYqeXGA4l",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "print(\"The shape of the dataset is {:d} rows by {:d} columns\".format(ratings_df.count(), len(ratings_df.columns)))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "-WC4RPQgyEsB",
        "colab_type": "text"
      },
      "source": [
        "### **Print out the schema of the loaded dataset**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Q3KhtSnvGIwG",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "ratings_df.printSchema()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dXlcRfzmNNCv",
        "colab_type": "text"
      },
      "source": [
        "### **Dataset Shape and Schema**\n",
        "\n",
        "The dataset contains **25,000,095** records, each one corresponding to the `rating` which an anonymized `userId` gave to a specific `movieId`. Moreover, the `timestamp` associated with each rating is also recorded.\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "0Pjib1fiylb6",
        "colab_type": "text"
      },
      "source": [
        "### **Display the first 5 rows of the dataset**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "a4v-Z92rGXoe",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "ratings_df.show(5)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1Df7cSAKmqQi",
        "colab_type": "text"
      },
      "source": [
        "### **Check for any missing values**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ubkWy6w6mtvG",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "for c in ratings_df.columns:\n",
        "  print(\"N. of missing values of column `{:s}` = {:d}\".format(c, ratings_df.where(col(c).isNull()).count()))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dkPiF7OWZW1s",
        "colab_type": "text"
      },
      "source": [
        "### **Check the number of unique users**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "e0gOhy6nZcP8",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "print(\"The number of unique users are: {:d}\".format(ratings_df.select(\"userId\").distinct().count()))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dFn16spbeulK",
        "colab_type": "text"
      },
      "source": [
        "### **Check the number of unique movies**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "UuUkIw89exs2",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "print(\"The number of unique movies are: {:d}\".format(ratings_df.select(\"movieId\").distinct().count()))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DHTM4wS2ztOJ",
        "colab_type": "text"
      },
      "source": [
        "# **2. Data Exploration**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "9TKWRYgl3jPi",
        "colab_type": "text"
      },
      "source": [
        "### **Summary of Descriptive Statistics**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "FvTv38i73pVh",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "ratings_df.describe().toPandas()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "CK1VjwrvN4BV",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# To access plotting libraries, we need to first transform our PySpark DataFrame into a Pandas DataFrame\n",
        "ratings_pdf = ratings_df.toPandas() "
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "okVR9377jPTf",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Set some default plotting configuration using seaborn properties\n",
        "sns.set_style(\"darkgrid\")\n",
        "sns.set_context(\"notebook\", rc={\"lines.linewidth\": 2, \n",
        "                                \"xtick.labelsize\":14, \n",
        "                                \"ytick.labelsize\":14,\n",
        "                                \"axes.labelsize\": 18,\n",
        "                                \"axes.titlesize\": 20,\n",
        "                                })"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "DlRGHd9WOr5i",
        "colab_type": "text"
      },
      "source": [
        "### **Analysis of Rating Distributions**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "8esOySR-Oh1Q",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "fig, ax = plt.subplots(1, 1, figsize=(8,8))\n",
        "\n",
        "ax = sns.countplot(x=\"rating\", data=ratings_pdf)\n",
        "#ax = sns.barplot(x=\"rating\", y=\"rating\", data=ratings_pdf, estimator=lambda x: len(x)/len(ratings_pdf) * 100)\n",
        "#ax.set_ylabel(\"Frequency (%)\")"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "colab_type": "text",
        "id": "xlUz0-TNb4Jo"
      },
      "source": [
        "### **Dataset Splitting: Training vs. Test Set**\n",
        "\n",
        "Before moving along with any preprocessing involving data transformations, we will split our dataset into **2** portions:\n",
        "- _training set_ (e.g., accounting for **80%** of the total number of instances);\n",
        "- _test set_ (e.g., accounting for the remaining **20%** of instances)"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "ZSzZLA9QcA_P",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Randomly split our original dataset `house_df` into 80÷20 for training and test, respectively\n",
        "train_df, test_df = ratings_df.randomSplit([0.8, 0.2], seed=RANDOM_SEED)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IRIWWJnJLo7g",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "print(\"Training set size: {:d} instances\".format(train_df.count()))\n",
        "print(\"Test set size: {:d} instances\".format(test_df.count()))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "Kh179KdCcS81",
        "colab_type": "text"
      },
      "source": [
        "### **Working on the Training Set only**\n",
        "\n",
        "From now on, we will be working on the training set portion only. The test set will come back into play when we evaluate our learned model."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "teSVf-WFzvs-",
        "colab_type": "text"
      },
      "source": [
        "# **Matrix Factorization using Alternating Least Squares (ALS)**\n",
        "\n",
        "We use ALS to factorize the user-movie matrix $R_{m \\times n}$ into the product of two lower rank matrices: the **user-factor matrix** $X_{m\\times d}$ and the **item-factor matrix** $W_{n\\times d}$, using the training set above. To do so, we use the blocked implementation of ALS (i.e., the `ALS` object) provided by the [PySpark API](https://spark.apache.org/docs/2.2.0/api/python/pyspark.ml.html?highlight=als#pyspark.ml.recommendation.ALS) within the package `pyspark.ml.recommendation`.\n",
        "\n",
        "The general approach is iterative. During each iteration, one of the factor matrices is held constant, while the other is solved for using least squares. The newly-solved factor matrix is then held constant while solving for the other factor matrix\n",
        "\n",
        "Among all the parameters that the API offers, the following deserve specific attention:\n",
        "\n",
        "- `rank` is the rank of the user and item factor matrices $X$ and $W$, i.e., the number of latent features $d$ (by deafult `rank=10`);\n",
        "- `maxIter` is the maximum number of iterations performed (by default `maxIter=10`);\n",
        "- `regParam` is the regularization parameter (by default `regParam=0.1`);\n",
        "- `implicitPrefs` is a boolean flag to switch between using only explicit or implicit feedback version of ALS (by default `implicitPrefs=false`);\n",
        "- `coldStartStrategy` indicates the strategy for dealing with unknown or new users/items at prediction time (i.e., cold-start). This may be useful in cross-validation or production scenarios, for handling users/items that the model have not seen in the training data. Supported values are `nan` (default) or `drop`;\n",
        "- `numUserBlocks`/`numItemBlocks` indicate the number of blocks to process in parallel (by default both are set to `10`, set those to `-1` to allow Spark to autoconfigure those).\n",
        "\n",
        "As it is always the case, the optimal values of the **hyperparameters** above should be tuned using a dedicated portion of the dataset (i.e., **validation set**) or by performing $k$**-fold cross validation**."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "I90uLYC_QlOc",
        "colab_type": "text"
      },
      "source": [
        "## **A First Matrix Factorization Model**"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "wIRYiv_ZRFZJ",
        "colab_type": "text"
      },
      "source": [
        "### **Model Training**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "XqITInLuIjpT",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.recommendation import ALS\n",
        "# Build the recommendation model using ALS on the training data\n",
        "als = ALS(maxIter=5, regParam=0.01, userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\", coldStartStrategy=\"drop\")\n",
        "model = als.fit(train_df)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "mdnxkKdfRKC0",
        "colab_type": "text"
      },
      "source": [
        "### **Making Rating Predictions on the Test Set**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "MjAftXeNRSdN",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "predictions =  model.transform(test_df)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "15-7xmtrRXDX",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "predictions.select([\"userId\", \"movieId\", \"rating\", \"prediction\"]).show()"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7IKvMhvwSZrB",
        "colab_type": "text"
      },
      "source": [
        "### **Measuring Model's Performance**\n",
        "\n",
        "We will use Root Mean Squared Error (RMSE) measured on the held-out portion to assess the quality of our movie recommender system."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "zHLu-AJhSYnk",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "from pyspark.ml.evaluation import RegressionEvaluator\n",
        "\n",
        "evaluator = RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\")\n",
        "rmse = evaluator.evaluate(predictions)\n",
        "\n",
        "print(\"Root Mean Squared Error = {:.5f}\".format(rmse))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "mKWC5LfR6ibL",
        "colab_type": "text"
      },
      "source": [
        "## **Tuning Hyperparameters**\n",
        "\n",
        "In the following, we try to summarize the whole pipeline making use also of $k$-fold cross validation to get a better estimate of the generalization performance of our matrix factorization model.\n",
        "\n",
        "More specifically, we will tune the three hyperparameters: `rank`, `regParam`, and `maxIter`. "
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "Ux02QmTg7_Ps",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# This function defines the general pipeline for logistic regression\n",
        "def matrix_factorization(train, k_fold=5):\n",
        "\n",
        "    from pyspark.ml.recommendation import ALS\n",
        "    from pyspark.ml.tuning import CrossValidator, ParamGridBuilder\n",
        "    from pyspark.ml.evaluation import RegressionEvaluator\n",
        "    from pyspark.ml import Pipeline\n",
        "\n",
        "    als = ALS(userCol=\"userId\", itemCol=\"movieId\", ratingCol=\"rating\", coldStartStrategy=\"drop\")\n",
        "\n",
        "    #pipeline = Pipeline(stages=stages)\n",
        "\n",
        "    # We use a ParamGridBuilder to construct a grid of parameters to search over.\n",
        "    # A CrossValidator requires an Estimator, a set of Estimator ParamMaps, and an Evaluator.\n",
        "    # We use a ParamGridBuilder to construct a grid of parameters to search over.\n",
        "    # With 2 values for als.rank, 2 values for als.regParam, and 1 value for als.maxIter,\n",
        "    # this grid will have 2 x 2 x 1 = 4 parameter settings for CrossValidator to choose from.\n",
        "    param_grid = ParamGridBuilder()\\\n",
        "    .addGrid(als.rank, [10, 25]) \\\n",
        "    .addGrid(als.regParam, [0.01, 0.1]) \\\n",
        "    .addGrid(als.maxIter, [10]) \\\n",
        "    .build()\n",
        "    \n",
        "    cross_val = CrossValidator(estimator=als, \n",
        "                               estimatorParamMaps=param_grid,\n",
        "                               evaluator=RegressionEvaluator(metricName=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\"),\n",
        "                               numFolds=k_fold,\n",
        "                               collectSubModels=True # this flag allows us to store ALL the models trained during k-fold cross validation\n",
        "                               )\n",
        "\n",
        "    # Run cross-validation, and choose the best set of parameters.\n",
        "    cv_model = cross_val.fit(train)\n",
        "\n",
        "    return cv_model"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "xjnj2jXeRSEJ",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "cv_model = matrix_factorization(train_df)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "IHvPkEyEBUOf",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# This function summarizes all the models trained during k-fold cross validation\n",
        "def summarize_all_models(cv_models):\n",
        "    for k, models in enumerate(cv_models):\n",
        "        print(\"*************** Fold #{:d} ***************\\n\".format(k+1))\n",
        "        for i, m in enumerate(models):\n",
        "            print(\"--- Model #{:d} out of {:d} ---\".format(i+1, len(models)))\n",
        "            print(\"\\tParameters: rank=[{:d}]\".format(m.rank))\n",
        "            print(\"\\tModel summary: {}\\n\".format(m))\n",
        "        print(\"***************************************\\n\")"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "m50kfzDTRhsd",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Call the function above|\n",
        "summarize_all_models(cv_model.subModels)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "jJyI7BSC8cmA",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "for i, avg_rmse in enumerate(cv_model.avgMetrics):\n",
        "    print(\"Avg. RMSE computed across k-fold cross validation for model setting #{:d}: {:.3f}\".format(i+1, avg_rmse))"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "sg_ockQBQjNX",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "print(\"Best model according to k-fold cross validation: rank=[{:d}]\".\n",
        "      format(cv_model.bestModel.rank)\n",
        "      )\n",
        "print(cv_model.bestModel)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "yadIWG4pIcrr",
        "colab_type": "text"
      },
      "source": [
        "### **Using the best model from $k$-fold cross validation to make predictions**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "b01W9KqbSz6c",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "# Make predictions on the test set (`cv_model` contains the best model according to the result of k-fold cross validation)\n",
        "# `test_df` will follow exactly the same pipeline defined above, and already fit to `train_df`\n",
        "test_predictions = cv_model.transform(test_df)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "2Tlh8G3STxGh",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "test_predictions.select(\"userId\", \"movieId\", \"rating\", \"prediction\").show(5)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "FOA-EKGxU16J",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "def evaluate_model(predictions, metric=\"rmse\", labelCol=\"rating\", predictionCol=\"prediction\"):\n",
        "    \n",
        "    from pyspark.ml.evaluation import RegressionEvaluator\n",
        "\n",
        "    evaluator = RegressionEvaluator(metricName=metric, labelCol=labelCol, predictionCol=predictionCol)\n",
        "\n",
        "    return evaluator.evaluate(predictions)"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "7ulP5vi_VpnP",
        "colab_type": "text"
      },
      "source": [
        "### **Evaluate model performance on the Test Set**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "WDEu2HuaaueC",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "print(\"***** Test Set *****\")\n",
        "print(\"RMSE: {:.3f}\".format(evaluate_model(test_predictions)))\n",
        "print(\"***** Test Set *****\")"
      ],
      "execution_count": 0,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "1qRVEXCk0U2q",
        "colab_type": "text"
      },
      "source": [
        "### **Provide top-$k$ Recommendations to all users**"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {
        "id": "lsME0ZKo0YKi",
        "colab_type": "code",
        "colab": {}
      },
      "source": [
        "k = 10 # number of recommended items for each user\n",
        "user_recs = cv_model.bestModel.recommendForAllUsers(K).show(10, truncate=False)"
      ],
      "execution_count": 0,
      "outputs": []
    }
  ]
}