notebooks/building-recommender.ipynb

{
 "metadata": {
  "name": "",
  "signature": "sha256:795b28fd3aeda16a4cb83908af89b3236003d3ff89d585e6e6daa76437f8ed01"
 },
 "nbformat": 3,
 "nbformat_minor": 0,
 "worksheets": [
  {
   "cells": [
    {
     "cell_type": "heading",
     "level": 1,
     "metadata": {},
     "source": [
      "An on-line movie recommending service using Spark & Flask - Building the recommender"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "This notebook explains how to use the [MovieLens dataset](http://grouplens.org/datasets/movielens/) to build a movie recommender using [collaborative filtering](https://en.wikipedia.org/wiki/Recommender_system#Collaborative_filtering) with [Spark's Alternating Least Saqures](https://spark.apache.org/docs/latest/mllib-collaborative-filtering.html) implementation. It is organised in two parts. The first one is about getting and parsing movies and ratings data into Spark RDDs. The second is about building and using the recommender and persisting it for later use in our on-line recommender system.    "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "This tutorial can be used independently to build a movie recommender model based on the MovieLens dataset. Most of the code in this first part, about how to use ALS with the public MovieLens dataset, comes from my solution to one of the exercises proposed in the [CS100.1x Introduction to Big Data with Apache Spark by Anthony D. Joseph on edX](https://www.edx.org/course/introduction-big-data-apache-spark-uc-berkeleyx-cs100-1x), that is also [**publicly available since 2014 at Spark Summit**](https://databricks-training.s3.amazonaws.com/movie-recommendation-with-mllib.html) . There I've added with minor modifications to use a larger dataset and also code about how to store and reload the model for later use. "
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Getting and processing the data"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In order to build an on-line movie recommender using Spark, we need to have our model data as preprocessed as possible. Parsing the dataset and building the model everytime a new recommendation needs to be done is not the best of the strategies."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The list of task we can pre-compute includes:  \n",
      "\n",
      "- Loading and parsing the dataset. Persisting the resulting RDD for later use.  \n",
      "- Building the recommender model using the complete dataset. Persist the dataset for later use.  \n",
      "\n",
      "This notebook explains the first of these tasks.  "
     ]
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "File download"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "GroupLens Research has collected and made available rating data sets from the [MovieLens web site](http://movielens.org). The data sets were collected over various periods of time, depending on the size of the set. They can be found [here](http://grouplens.org/datasets/movielens/).   "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In our case, we will use the latest datasets:  \n",
      "\n",
      "- Small: 100,000 ratings and 2,488 tag applications applied to 8,570 movies by 706 users. Last updated 4/2015.  \n",
      "- Full: 21,000,000 ratings and 470,000 tag applications applied to 27,000 movies by 230,000 users. Last updated 4/2015.  \n"
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "complete_dataset_url = 'http://files.grouplens.org/datasets/movielens/ml-latest.zip'\n",
      "small_dataset_url = 'http://files.grouplens.org/datasets/movielens/ml-latest-small.zip'"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 1
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We also need to define download locations."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import os\n",
      "\n",
      "datasets_path = os.path.join('..', 'datasets')\n",
      "\n",
      "complete_dataset_path = os.path.join(datasets_path, 'ml-latest.zip')\n",
      "small_dataset_path = os.path.join(datasets_path, 'ml-latest-small.zip')"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 2
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we can proceed with both downloads."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import urllib\n",
      "\n",
      "small_f = urllib.urlretrieve (small_dataset_url, small_dataset_path)\n",
      "complete_f = urllib.urlretrieve (complete_dataset_url, complete_dataset_path)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Both of them are zip files containing a folder with ratings, movies, etc. We need to extract them into its individual folders so we can use each file later on.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "import zipfile\n",
      "\n",
      "with zipfile.ZipFile(small_dataset_path, \"r\") as z:\n",
      "    z.extractall(datasets_path)\n",
      "\n",
      "with zipfile.ZipFile(complete_dataset_path, \"r\") as z:\n",
      "    z.extractall(datasets_path)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 4
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "Loading and parsing datasets"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "No we are ready to read in each of the files and create an RDD consisting of parsed lines.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Each line in the ratings dataset (`ratings.csv`) is formatted as:  \n",
      "\n",
      "`userId,movieId,rating,timestamp`  \n",
      "\n",
      "Each line in the movies (`movies.csv`) dataset is formatted as:  \n",
      "\n",
      "`movieId,title,genres`  \n",
      "\n",
      "Were *genres* has the format:  \n",
      "\n",
      "`Genre1|Genre2|Genre3...`\n",
      "\n",
      "The tags file (`tags.csv`) has the format:  \n",
      "\n",
      "`userId,movieId,tag,timestamp`  \n",
      "\n",
      "And finally, the `links.csv` file has the format:  \n",
      "\n",
      "`movieId,imdbId,tmdbId`  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The format of these files is uniform and simple, so we can use Python [`split()`](https://docs.python.org/2/library/stdtypes.html#str.split) to parse their lines once they are loaded into RDDs. Parsing the movies and ratings files yields two RDDs:  \n",
      "\n",
      "* For each line in the ratings dataset, we create a tuple of `(UserID, MovieID, Rating)`. We drop the *timestamp* because we do not need it for this recommender.  \n",
      "* For each line in the movies dataset, we create a tuple of `(MovieID, Title)`. We drop the *genres* because we do not use them for this recommender.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "So let's load the raw ratings data. We need to filter out the header, included in each file.    "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "small_ratings_file = os.path.join(datasets_path, 'ml-latest-small', 'ratings.csv')\n",
      "\n",
      "small_ratings_raw_data = sc.textFile(small_ratings_file)\n",
      "small_ratings_raw_data_header = small_ratings_raw_data.take(1)[0]"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 3
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we can parse the raw data into a new RDD.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "small_ratings_data = small_ratings_raw_data.filter(lambda line: line!=small_ratings_raw_data_header)\\\n",
      "    .map(lambda line: line.split(\",\")).map(lambda tokens: (tokens[0],tokens[1],tokens[2])).cache()"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 4
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "For illustrative purposes, we can take the first few lines of our RDD to see the result. In the final script we don't call any Spark action (e.g. `take`) until needed, since they trigger actual computations in the cluster.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "small_ratings_data.take(3)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 5,
       "text": [
        "[(u'1', u'6', u'2.0'), (u'1', u'22', u'3.0'), (u'1', u'32', u'2.0')]"
       ]
      }
     ],
     "prompt_number": 5
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We proceed in a similar way with the `movies.csv` file."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "small_movies_file = os.path.join(datasets_path, 'ml-latest-small', 'movies.csv')\n",
      "\n",
      "small_movies_raw_data = sc.textFile(small_movies_file)\n",
      "small_movies_raw_data_header = small_movies_raw_data.take(1)[0]\n",
      "\n",
      "small_movies_data = small_movies_raw_data.filter(lambda line: line!=small_movies_raw_data_header)\\\n",
      "    .map(lambda line: line.split(\",\")).map(lambda tokens: (tokens[0],tokens[1])).cache()\n",
      "    \n",
      "small_movies_data.take(3)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 6,
       "text": [
        "[(u'1', u'Toy Story (1995)'),\n",
        " (u'2', u'Jumanji (1995)'),\n",
        " (u'3', u'Grumpier Old Men (1995)')]"
       ]
      }
     ],
     "prompt_number": 6
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The following sections introduce *Collaborative Filtering* and explain how to use *Spark MLlib* to build a recommender model. We will close the tutorial by explaining how a model such this is used to make recommendations, and how to persist it for later use (e.g. in our Python/flask web-service)."
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Collaborative Filtering"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In Collaborative filtering we make predictions (filtering) about the interests of a user by collecting preferences or taste information from many users (collaborating). The underlying assumption is that if a user A has the same opinion as a user B on an issue, A is more likely to have B's opinion on a different issue x than to have the opinion on x of a user chosen randomly.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "The image below (from [Wikipedia](https://en.wikipedia.org/?title=Collaborative_filtering)) shows an example of collaborative filtering. At first, people rate different items (like videos, images, games). Then, the system makes predictions about a user's rating for an item not rated yet. The new predictions are built upon the existing ratings of other users with similar ratings with the active user. In the image, the system predicts that the user will not like the video.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "![collaborative filtering](https://upload.wikimedia.org/wikipedia/commons/5/52/Collaborative_filtering.gif)"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Spark MLlib library for Machine Learning provides a [Collaborative Filtering](https://spark.apache.org/docs/latest/mllib-collaborative-filtering.html) implementation by using [Alternating Least Squares](http://dl.acm.org/citation.cfm?id=1608614). The implementation in MLlib has the following parameters:  \n",
      "\n",
      "- numBlocks is the number of blocks used to parallelize computation (set to -1 to auto-configure).  \n",
      "- rank is the number of latent factors in the model.  \n",
      "- iterations is the number of iterations to run.  \n",
      "- lambda specifies the regularization parameter in ALS.  \n",
      "- implicitPrefs specifies whether to use the explicit feedback ALS variant or one adapted for implicit feedback data.  \n",
      "- alpha is a parameter applicable to the implicit feedback variant of ALS that governs the baseline confidence in preference observations.  \n"
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Selecting ALS parameters using the small dataset"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In order to determine the best ALS parameters, we will use the small dataset. We need first to split it into train, validation, and test datasets."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "training_RDD, validation_RDD, test_RDD = small_ratings_data.randomSplit([6, 2, 2], seed=0L)\n",
      "validation_for_predict_RDD = validation_RDD.map(lambda x: (x[0], x[1]))\n",
      "test_for_predict_RDD = test_RDD.map(lambda x: (x[0], x[1]))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 7
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we can proceed with the training phase. "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from pyspark.mllib.recommendation import ALS\n",
      "import math\n",
      "\n",
      "seed = 5L\n",
      "iterations = 10\n",
      "regularization_parameter = 0.1\n",
      "ranks = [4, 8, 12]\n",
      "errors = [0, 0, 0]\n",
      "err = 0\n",
      "tolerance = 0.02\n",
      "\n",
      "min_error = float('inf')\n",
      "best_rank = -1\n",
      "best_iteration = -1\n",
      "for rank in ranks:\n",
      "    model = ALS.train(training_RDD, rank, seed=seed, iterations=iterations,\n",
      "                      lambda_=regularization_parameter)\n",
      "    predictions = model.predictAll(validation_for_predict_RDD).map(lambda r: ((r[0], r[1]), r[2]))\n",
      "    rates_and_preds = validation_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predictions)\n",
      "    error = math.sqrt(rates_and_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean())\n",
      "    errors[err] = error\n",
      "    err += 1\n",
      "    print 'For rank %s the RMSE is %s' % (rank, error)\n",
      "    if error < min_error:\n",
      "        min_error = error\n",
      "        best_rank = rank\n",
      "\n",
      "print 'The best model was trained with rank %s' % best_rank"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "For rank 4 the RMSE is 0.963681878574\n",
        "For rank 8 the RMSE is 0.96250475933"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "For rank 12 the RMSE is 0.971647563632"
       ]
      },
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "\n",
        "The best model was trained with rank 8\n"
       ]
      }
     ],
     "prompt_number": 8
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "But let's explain this a little bit. First, let's have a look at how our predictions look.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "predictions.take(3)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 9,
       "text": [
        "[((32, 4018), 3.280114696166238),\n",
        " ((375, 4018), 2.7365714977314086),\n",
        " ((674, 4018), 2.510684514310653)]"
       ]
      }
     ],
     "prompt_number": 9
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Basically we have the UserID, the MovieID, and the Rating, as we have in our ratings dataset. In this case the predictions third element, the rating for that movie and user, is the predicted by our ALS model."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Then we join these with our validation data (the one that includes ratings) and the result looks as follows:  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "rates_and_preds.take(3)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 10,
       "text": [
        "[((558, 788), (3.0, 3.0419325487471403)),\n",
        " ((176, 3550), (4.5, 3.3214065001580986)),\n",
        " ((302, 3908), (1.0, 2.4728711204440765))]"
       ]
      }
     ],
     "prompt_number": 10
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "To that, we apply a squared difference and the we use the `mean()` action to get the MSE and apply `sqrt`."
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Finally we test the selected model."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "model = ALS.train(training_RDD, best_rank, seed=seed, iterations=iterations,\n",
      "                      lambda_=regularization_parameter)\n",
      "predictions = model.predictAll(test_for_predict_RDD).map(lambda r: ((r[0], r[1]), r[2]))\n",
      "rates_and_preds = test_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predictions)\n",
      "error = math.sqrt(rates_and_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean())\n",
      "    \n",
      "print 'For testing data the RMSE is %s' % (error)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "For testing data the RMSE is 0.972342381898\n"
       ]
      }
     ],
     "prompt_number": 11
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Using the complete dataset to build the final model"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "In order to build our recommender model, we will use the complete dataset. Therefore, we need to process it the same way we did with the small dataset.   "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Load the complete dataset file\n",
      "complete_ratings_file = os.path.join(datasets_path, 'ml-latest', 'ratings.csv')\n",
      "complete_ratings_raw_data = sc.textFile(complete_ratings_file)\n",
      "complete_ratings_raw_data_header = complete_ratings_raw_data.take(1)[0]\n",
      "\n",
      "# Parse\n",
      "complete_ratings_data = complete_ratings_raw_data.filter(lambda line: line!=complete_ratings_raw_data_header)\\\n",
      "    .map(lambda line: line.split(\",\")).map(lambda tokens: (int(tokens[0]),int(tokens[1]),float(tokens[2]))).cache()\n",
      "    \n",
      "print \"There are %s recommendations in the complete dataset\" % (complete_ratings_data.count())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "There are 21063128 recommendations in the complete dataset\n"
       ]
      }
     ],
     "prompt_number": 12
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we are ready to train the recommender model."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "training_RDD, test_RDD = complete_ratings_data.randomSplit([7, 3], seed=0L)\n",
      "\n",
      "complete_model = ALS.train(training_RDD, best_rank, seed=seed, \n",
      "                           iterations=iterations, lambda_=regularization_parameter)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 13
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we test on our testing set.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "test_for_predict_RDD = test_RDD.map(lambda x: (x[0], x[1]))\n",
      "\n",
      "predictions = complete_model.predictAll(test_for_predict_RDD).map(lambda r: ((r[0], r[1]), r[2]))\n",
      "rates_and_preds = test_RDD.map(lambda r: ((int(r[0]), int(r[1])), float(r[2]))).join(predictions)\n",
      "error = math.sqrt(rates_and_preds.map(lambda r: (r[1][0] - r[1][1])**2).mean())\n",
      "    \n",
      "print 'For testing data the RMSE is %s' % (error)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "For testing data the RMSE is 0.82183583368\n"
       ]
      }
     ],
     "prompt_number": 14
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We can see how we got a more accurate recommender when using a much larger dataset.  "
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "How to make recommendations"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Although we aim at building an on-line movie recommender, now that we know how to have our recommender model ready, we can give it a try providing some movie recommendations. This will help us coiding the recommending engine later on when building the web service, and will explain how to use the model in any other circumstances.  "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "When using collaborative filtering, getting recommendations is not as simple as predicting for the new entries using a previously generated model. Instead, we need to train again the model but including the new user preferences in order to compare them with other users in the dataset. That is, the recommender needs to be trained every time we have new user ratings (although a single model can be used by multiple users of course!). This makes the process expensive, and it is one of the reasons why scalability is a problem (and Spark a solution!). Once we have our model trained, we can reuse it to obtain top recomendations for a given user or an individual rating for a particular movie. These are less costly operations than training the model itself.    "
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "So let's first load the movies complete file for later use."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "complete_movies_file = os.path.join(datasets_path, 'ml-latest', 'movies.csv')\n",
      "complete_movies_raw_data = sc.textFile(complete_movies_file)\n",
      "complete_movies_raw_data_header = complete_movies_raw_data.take(1)[0]\n",
      "\n",
      "# Parse\n",
      "complete_movies_data = complete_movies_raw_data.filter(lambda line: line!=complete_movies_raw_data_header)\\\n",
      "    .map(lambda line: line.split(\",\")).map(lambda tokens: (int(tokens[0]),tokens[1],tokens[2])).cache()\n",
      "\n",
      "complete_movies_titles = complete_movies_data.map(lambda x: (int(x[0]),x[1]))\n",
      "    \n",
      "print \"There are %s movies in the complete dataset\" % (complete_movies_titles.count())"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "There are 27303 movies in the complete dataset\n"
       ]
      }
     ],
     "prompt_number": 15
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Another thing we want to do, is give recommendations of movies with a certain minimum number of ratings. For that, we need to count the number of ratings per movie.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "def get_counts_and_averages(ID_and_ratings_tuple):\n",
      "    nratings = len(ID_and_ratings_tuple[1])\n",
      "    return ID_and_ratings_tuple[0], (nratings, float(sum(x for x in ID_and_ratings_tuple[1]))/nratings)\n",
      "\n",
      "movie_ID_with_ratings_RDD = (complete_ratings_data.map(lambda x: (x[1], x[2])).groupByKey())\n",
      "movie_ID_with_avg_ratings_RDD = movie_ID_with_ratings_RDD.map(get_counts_and_averages)\n",
      "movie_rating_counts_RDD = movie_ID_with_avg_ratings_RDD.map(lambda x: (x[0], x[1][0]))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 17
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "Adding new user ratings"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we need to rate some movies for the new user. We will put them in a new RDD and we will use the user ID 0, that is not assigned in the MovieLens dataset. Check the [dataset](http://grouplens.org/datasets/movielens/) movies file for ID to Tittle assignment (so you know what movies are you actually rating).   "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "new_user_ID = 0\n",
      "\n",
      "# The format of each line is (userID, movieID, rating)\n",
      "new_user_ratings = [\n",
      "     (0,260,9), # Star Wars (1977)\n",
      "     (0,1,8), # Toy Story (1995)\n",
      "     (0,16,7), # Casino (1995)\n",
      "     (0,25,8), # Leaving Las Vegas (1995)\n",
      "     (0,32,9), # Twelve Monkeys (a.k.a. 12 Monkeys) (1995)\n",
      "     (0,335,4), # Flintstones, The (1994)\n",
      "     (0,379,3), # Timecop (1994)\n",
      "     (0,296,7), # Pulp Fiction (1994)\n",
      "     (0,858,10) , # Godfather, The (1972)\n",
      "     (0,50,8) # Usual Suspects, The (1995)\n",
      "    ]\n",
      "new_user_ratings_RDD = sc.parallelize(new_user_ratings)\n",
      "print 'New user ratings: %s' % new_user_ratings_RDD.take(10)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "New user ratings: [(0, 260, 9), (0, 1, 8), (0, 16, 7), (0, 25, 8), (0, 32, 9), (0, 335, 4), (0, 379, 3), (0, 296, 7), (0, 858, 10), (0, 50, 8)]\n"
       ]
      }
     ],
     "prompt_number": 18
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Now we add them to the data we will use to train our recommender model. We use Spark's `union()` transformation for this.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "complete_data_with_new_ratings_RDD = complete_ratings_data.union(new_user_ratings_RDD)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 19
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "And finally we train the ALS model using all the parameters we selected before (when using the small dataset)."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from time import time\n",
      "\n",
      "t0 = time()\n",
      "new_ratings_model = ALS.train(complete_data_with_new_ratings_RDD, best_rank, seed=seed, \n",
      "                              iterations=iterations, lambda_=regularization_parameter)\n",
      "tt = time() - t0\n",
      "\n",
      "print \"New model trained in %s seconds\" % round(tt,3)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "New model trained in 56.61 seconds\n"
       ]
      }
     ],
     "prompt_number": 20
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "It took some time. We will need to repeat that every time a user add new ratings. Ideally we will do this in batches, and not for every single rating that comes into the system for every user."
     ]
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "Getting top recommendations"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Let's now get some recommendations! For that we will get an RDD with all the movies the new user hasn't rated yet. We will them together with the model to predict ratings.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
     "new_user_ratings_ids = map(lambda x: x[1], new_user_ratings) # get just movie IDs\n",
     "# keep just those not on the ID list (thanks Lei Li for spotting the error!)\n",
     "new_user_unrated_movies_RDD = (complete_movies_data.filter(lambda x: x[0] not in new_user_ratings_ids).map(lambda x: (new_user_ID, x[0])))\n",
      "\n",
      "# Use the input RDD, new_user_unrated_movies_RDD, with new_ratings_model.predictAll() to predict new ratings for the movies\n",
      "new_user_recommendations_RDD = new_ratings_model.predictAll(new_user_unrated_movies_RDD)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 21
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We have our recommendations ready. Now we can print out the 25 movies with the highest predicted ratings. And join them with the movies RDD to get the titles, and ratings count in order to get movies with a minimum number of counts. First we will do the join and see what does the result looks like."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "# Transform new_user_recommendations_RDD into pairs of the form (Movie ID, Predicted Rating)\n",
      "new_user_recommendations_rating_RDD = new_user_recommendations_RDD.map(lambda x: (x.product, x.rating))\n",
      "new_user_recommendations_rating_title_and_count_RDD = \\\n",
      "    new_user_recommendations_rating_RDD.join(complete_movies_titles).join(movie_rating_counts_RDD)\n",
      "new_user_recommendations_rating_title_and_count_RDD.take(3)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 22,
       "text": [
        "[(87040, ((6.834512984654888, u'\"Housemaid'), 14)),\n",
        " (8194, ((5.966704041954459, u'Baby Doll (1956)'), 79)),\n",
        " (130390, ((0.6922328127396398, u'Contract Killers (2009)'), 1))]"
       ]
      }
     ],
     "prompt_number": 22
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "So we need to flat this down a bit in order to have `(Title, Rating, Ratings Count)`."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "new_user_recommendations_rating_title_and_count_RDD = \\\n",
      "    new_user_recommendations_rating_title_and_count_RDD.map(lambda r: (r[1][0][1], r[1][0][0], r[1][1]))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 23
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Finally, get the highest rated recommendations for the new user, filtering out movies with less than 25 ratings."
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "top_movies = new_user_recommendations_rating_title_and_count_RDD.filter(lambda r: r[2]>=25).takeOrdered(25, key=lambda x: -x[1])\n",
      "\n",
      "print ('TOP recommended movies (with more than 25 reviews):\\n%s' %\n",
      "        '\\n'.join(map(str, top_movies)))"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "output_type": "stream",
       "stream": "stdout",
       "text": [
        "TOP recommended movies (with more than 25 reviews):\n",
        "(u'\"Godfather: Part II', 8.503749129186701, 29198)\n",
        "(u'\"Civil War', 8.386497469089297, 257)\n",
        "(u'Frozen Planet (2011)', 8.372705479107108, 31)\n",
        "(u'\"Shawshank Redemption', 8.258510064442426, 67741)\n",
        "(u'Cosmos (1980)', 8.252254825768972, 948)\n",
        "(u'Band of Brothers (2001)', 8.225114960311624, 4450)\n",
        "(u'Generation Kill (2008)', 8.206487040524653, 52)\n",
        "(u\"Schindler's List (1993)\", 8.172761674773625, 53609)\n",
        "(u'Dr. Strangelove or: How I Learned to Stop Worrying and Love the Bomb (1964)', 8.166229786764168, 23915)\n",
        "(u\"One Flew Over the Cuckoo's Nest (1975)\", 8.15617022970577, 32948)\n",
        "(u'Casablanca (1942)', 8.141303207981174, 26114)\n",
        "(u'Seven Samurai (Shichinin no samurai) (1954)', 8.139633165142612, 11796)\n",
        "(u'Goodfellas (1990)', 8.12931139039048, 27123)\n",
        "(u'Star Wars: Episode V - The Empire Strikes Back (1980)', 8.124225700242096, 47710)\n",
        "(u'Jazz (2001)', 8.078538221315313, 25)\n",
        "(u\"Long Night's Journey Into Day (2000)\", 8.050176820606127, 34)\n",
        "(u'Lawrence of Arabia (1962)', 8.041331489948814, 13452)\n",
        "(u'Raiders of the Lost Ark (Indiana Jones and the Raiders of the Lost Ark) (1981)', 8.0399424815528, 45908)\n",
        "(u'12 Angry Men (1957)', 8.011389274280754, 13235)\n",
        "(u\"It's Such a Beautiful Day (2012)\", 8.007734839026181, 35)\n",
        "(u'Apocalypse Now (1979)', 8.005094327199552, 23905)\n",
        "(u'Paths of Glory (1957)', 7.999379786394267, 3598)\n",
        "(u'Rear Window (1954)', 7.9860865203540214, 17996)\n",
        "(u'State of Play (2003)', 7.981582126801772, 27)\n",
        "(u'Chinatown (1974)', 7.978673289692703, 16195)\n"
       ]
      }
     ],
     "prompt_number": 24
    },
    {
     "cell_type": "heading",
     "level": 3,
     "metadata": {},
     "source": [
      "Getting individual ratings"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Another useful usecase is getting the predicted rating for a particular movie for a given user. The process is similar to the previous retreival of top recommendations but, instead of using `predcitAll` with every single movie the user hasn't rated yet, we will just pass the method a single entry with the movie we want to predict the rating for.  "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "my_movie = sc.parallelize([(0, 500)]) # Quiz Show (1994)\n",
      "individual_movie_rating_RDD = new_ratings_model.predictAll(new_user_unrated_movies_RDD)\n",
      "individual_movie_rating_RDD.take(1)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [
      {
       "metadata": {},
       "output_type": "pyout",
       "prompt_number": 25,
       "text": [
        "[Rating(user=0, product=122880, rating=4.955831875971526)]"
       ]
      }
     ],
     "prompt_number": 25
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Not very likely that the new user will like that one... Obviously we can include as many movies as we need in that list!"
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Persisting the model"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Optionally, we might want to persist the base model for later use in our on-line recommendations. Although a new model is generated everytime we have new user ratings, it might be worth it to store the current one, in order to save time when starting up the server, etc. We might also save time if we persist some of the RDDs we have generated, specially those that took longer to process. For example, the following lines save and load a ALS model.      "
     ]
    },
    {
     "cell_type": "code",
     "collapsed": false,
     "input": [
      "from pyspark.mllib.recommendation import MatrixFactorizationModel\n",
      "\n",
      "model_path = os.path.join('..', 'models', 'movie_lens_als')\n",
      "\n",
      "# Save and load model\n",
      "model.save(sc, model_path)\n",
      "same_model = MatrixFactorizationModel.load(sc, model_path)"
     ],
     "language": "python",
     "metadata": {},
     "outputs": [],
     "prompt_number": 27
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "Among other things, you will see in your filesystem that there are folder with product and user data into [Parquet](https://parquet.apache.org/) format files.  "
     ]
    },
    {
     "cell_type": "heading",
     "level": 2,
     "metadata": {},
     "source": [
      "Genre and other fields"
     ]
    },
    {
     "cell_type": "markdown",
     "metadata": {},
     "source": [
      "We havent used the `genre` and `timestamp` fields in order to simplify the transformations and the whole tutorial. Incorporating them doesn't reprensent any problem. A good use could be filtering recommendations by any of them (e.g. recommendations by genre, or recent recommendations) like we have done with the minimum number of ratings.  "
     ]
    }
   ],
   "metadata": {}
  }
 ]
}