cs110_lab2_als_prediction.ipynb

{"cells":[{"cell_type":"markdown","source":["<a rel=\"license\" href=\"http://creativecommons.org/licenses/by-nc-nd/4.0/\"> <img alt=\"Creative Commons License\" style=\"border-width:0\" src=\"https://i.creativecommons.org/l/by-nc-nd/4.0/88x31.png\"/> </a> <br/> This work is licensed under a <a rel=\"license\" href=\"http://creativecommons.org/licenses/by-nc-nd/4.0/\"> Creative Commons Attribution-NonCommercial-NoDerivatives 4.0 International License. </a>"],"metadata":{}},{"cell_type":"markdown","source":["#![Spark Logo](http://spark-mooc.github.io/web-assets/images/ta_Spark-logo-small.png) + ![Python Logo](http://spark-mooc.github.io/web-assets/images/python-logo-master-v3-TM-flattened_small.png)\n\n<img src=\"http://spark-mooc.github.io/web-assets/images/cs110x/movie-camera.png\" style=\"float:right; height: 200px; margin: 10px; border: 1px solid #ddd; border-radius: 15px 15px 15px 15px; padding: 10px\"/>\n\n# Predicting Movie Ratings\n\nOne of the most common uses of big data is to predict what users want.  This allows Google to show you relevant ads, Amazon to recommend relevant products, and Netflix to recommend movies that you might like.  This lab will demonstrate how we can use Apache Spark to recommend movies to a user.  We will start with some basic techniques, and then use the [Spark ML][sparkml] library's Alternating Least Squares method to make more sophisticated predictions.\n\nFor this lab, we will use a subset dataset of 20 million ratings. This dataset is pre-mounted on Databricks and is from the [MovieLens stable benchmark rating dataset](http://grouplens.org/datasets/movielens/). However, the same code you write will also work on the full dataset (though running with the full dataset on Community Edition is likely to take quite a long time).\n\nIn this lab:\n* *Part 0*: Preliminaries\n* *Part 1*: Basic Recommendations\n* *Part 2*: Collaborative Filtering\n* *Part 3*: Predictions for Yourself\n\nAs mentioned during the first Learning Spark lab, think carefully before calling `collect()` on any datasets.  When you are using a small dataset, calling `collect()` and then using Python to get a sense for the data locally (in the driver program) will work fine, but this will not work when you are using a large dataset that doesn't fit in memory on one machine.  Solutions that call `collect()` and do local analysis that could have been done with Spark will likely fail in the autograder and not receive full credit.\n[sparkml]: https://spark.apache.org/docs/1.6.2/api/python/pyspark.ml.html"],"metadata":{}},{"cell_type":"code","source":["labVersion = 'cs110x.lab2-1.0.0'"],"metadata":{},"outputs":[],"execution_count":3},{"cell_type":"markdown","source":["## Code\n\nThis assignment can be completed using basic Python and pySpark DataFrame Transformations and Actions.  Libraries other than math are not necessary. With the exception of the ML functions that we introduce in this assignment, you should be able to complete all parts of this homework using only the Spark functions you have used in prior lab exercises (although you are welcome to use more features of Spark if you like!).\n\nWe'll be using motion picture data, the same data last year's CS100.1x used. However, in this course, we're using DataFrames, rather than RDDs.\n\nThe following cell defines the locations of the data files. If you want to run an exported version of this lab on your own machine (i.e., outside of Databricks), you'll need to download your own copy of the 20-million movie data set, and you'll need to adjust the paths, below.\n\n**To Do**: Run the following cell."],"metadata":{}},{"cell_type":"code","source":["import os\nfrom databricks_test_helper import Test\n\ndbfs_dir = '/databricks-datasets/cs110x/ml-20m/data-001'\nratings_filename = dbfs_dir + '/ratings.csv'\nmovies_filename = dbfs_dir + '/movies.csv'\n\n# The following line is here to enable this notebook to be exported as source and\n# run on a local machine with a local copy of the files. Just change the dbfs_dir,\n# above.\nif os.path.sep != '/':\n  # Handle Windows.\n  ratings_filename = ratings_filename.replace('/', os.path.sep)\n  movie_filename = movie_filename.replace('/', os.path.sep)"],"metadata":{},"outputs":[],"execution_count":5},{"cell_type":"markdown","source":["## Part 0: Preliminaries\n\nWe read in each of the files and create a DataFrame consisting of parsed lines.\n\n### The 20-million movie sample\n\nThe 20-million movie sample consists of CSV files (with headers), so there's no need to parse the files manually, as Spark CSV can do the job."],"metadata":{}},{"cell_type":"markdown","source":["First, let's take a look at the directory containing our files."],"metadata":{}},{"cell_type":"code","source":["display(dbutils.fs.ls(dbfs_dir))"],"metadata":{},"outputs":[],"execution_count":8},{"cell_type":"markdown","source":["### CPU vs I/O tradeoff\n\nNote that we have both compressed files (ending in `.gz`) and uncompressed files. We have a CPU vs. I/O tradeoff here. If I/O is the bottleneck, then we want to process the compressed files and pay the extra CPU overhead. If CPU is the bottleneck, then it makes more sense to process the uncompressed files.\n\nWe've done some experiments, and we've determined that CPU is more of a bottleneck than I/O, on Community Edition. So, we're going to process the uncompressed data. In addition, we're going to speed things up further by specifying the DataFrame schema explicitly. (When the Spark CSV adapter infers the schema from a CSV file, it has to make an extra pass over the file. That'll slow things down here, and it isn't really necessary.)\n\n**To Do**: Run the following cell, which will define the schemas."],"metadata":{}},{"cell_type":"code","source":["from pyspark.sql.types import *\n\nratings_df_schema = StructType(\n  [StructField('userId', IntegerType()),\n   StructField('movieId', IntegerType()),\n   StructField('rating', DoubleType())]\n)\nmovies_df_schema = StructType(\n  [StructField('ID', IntegerType()),\n   StructField('title', StringType())]\n)"],"metadata":{},"outputs":[],"execution_count":10},{"cell_type":"markdown","source":["### Load and Cache\n\nThe Databricks File System (DBFS) sits on top of S3. We're going to be accessing this data a lot. Rather than read it over and over again from S3, we'll cache both\nthe movies DataFrame and the ratings DataFrame in memory.\n\n**To Do**: Run the following cell to load and cache the data. Please be patient: The code takes about 30 seconds to run."],"metadata":{}},{"cell_type":"code","source":["from pyspark.sql.functions import regexp_extract\nfrom pyspark.sql.types import *\n\nraw_ratings_df = sqlContext.read.format('com.databricks.spark.csv').options(header=True, inferSchema=False).schema(ratings_df_schema).load(ratings_filename)\nratings_df = raw_ratings_df.drop('Timestamp')\n\nraw_movies_df = sqlContext.read.format('com.databricks.spark.csv').options(header=True, inferSchema=False).schema(movies_df_schema).load(movies_filename)\nmovies_df = raw_movies_df.drop('Genres').withColumnRenamed('movieId', 'ID')\n\nratings_df.cache()\nmovies_df.cache()\n\nassert ratings_df.is_cached\nassert movies_df.is_cached\n\nraw_ratings_count = raw_ratings_df.count()\nratings_count = ratings_df.count()\nraw_movies_count = raw_movies_df.count()\nmovies_count = movies_df.count()\n\nprint 'There are %s ratings and %s movies in the datasets' % (ratings_count, movies_count)\nprint 'Ratings:'\nratings_df.show(3)\nprint 'Movies:'\nmovies_df.show(3, truncate=False)\n\nassert raw_ratings_count == ratings_count\nassert raw_movies_count == movies_count"],"metadata":{},"outputs":[],"execution_count":12},{"cell_type":"markdown","source":["Next, let's do a quick verification of the data.\n\n**To do**: Run the following cell. It should run without errors."],"metadata":{}},{"cell_type":"code","source":["assert ratings_count == 20000263\nassert movies_count == 27278\nassert movies_df.filter(movies_df.title == 'Toy Story (1995)').count() == 1\nassert ratings_df.filter((ratings_df.userId == 6) & (ratings_df.movieId == 1) & (ratings_df.rating == 5.0)).count() == 1"],"metadata":{},"outputs":[],"execution_count":14},{"cell_type":"markdown","source":["Let's take a quick look at some of the data in the two DataFrames.\n\n**To Do**: Run the following two cells."],"metadata":{}},{"cell_type":"code","source":["display(movies_df)"],"metadata":{},"outputs":[],"execution_count":16},{"cell_type":"code","source":["display(ratings_df)"],"metadata":{},"outputs":[],"execution_count":17},{"cell_type":"markdown","source":["## Part 1: Basic Recommendations\n\nOne way to recommend movies is to always recommend the movies with the highest average rating. In this part, we will use Spark to find the name, number of ratings, and the average rating of the 20 movies with the highest average rating and at least 500 reviews. We want to filter our movies with high ratings but greater than or equal to 500 reviews because movies with few reviews may not have broad appeal to everyone."],"metadata":{}},{"cell_type":"markdown","source":["### (1a) Movies with Highest Average Ratings\n\nNow that we have a way to calculate the average ratings, we will use the `getCountsAndAverages()` helper function with Spark to determine the movies with the highest average ratings.\n\nThe steps you should perform are:\n\n1. Recall that the `ratings_df` contains three columns:\n    - The ID of the user who rated the film\n    - the ID of the movie being rated\n    - and the rating.\n\n   First, transform `ratings_df` into a second DataFrame, `movie_ids_with_avg_ratings`, with the following columns:\n    - The movie ID\n    - The number of ratings for the movie\n    - The average of all the movie's ratings\n\n2. Transform `movie_ids_with_avg_ratings` to another DataFrame, `movie_names_with_avg_ratings_df` that adds the movie name to each row. `movie_names_with_avg_ratings_df`\n   will contain these columns:\n    - The movie ID\n    - The movie name\n    - The number of ratings for the movie\n    - The average of all the movie's ratings\n\n   **Hint**: You'll need to do a join.\n\nYou should end up with something like the following:\n```\nmovie_ids_with_avg_ratings_df:\n+-------+-----+------------------+\n|movieId|count|average           |\n+-------+-----+------------------+\n|1831   |7463 |2.5785207021305103|\n|431    |8946 |3.695059244355019 |\n|631    |2193 |2.7273141814865483|\n+-------+-----+------------------+\nonly showing top 3 rows\n\nmovie_names_with_avg_ratings_df:\n+-------+-----------------------------+-----+-------+\n|average|title                        |count|movieId|\n+-------+-----------------------------+-----+-------+\n|5.0    |Ella Lola, a la Trilby (1898)|1    |94431  |\n|5.0    |Serving Life (2011)          |1    |129034 |\n|5.0    |Diplomatic Immunity (2009? ) |1    |107434 |\n+-------+-----------------------------+-----+-------+\nonly showing top 3 rows\n```"],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL_IN> with appropriate code\nfrom pyspark.sql import functions as F\n# From ratingsDF, create a movie_ids_with_avg_ratings_df that combines the two DataFrames\nmovie_ids_with_avg_ratings_df = ratings_df.groupBy('movieId').agg(F.count(ratings_df.rating).alias(\"count\"), F.avg(ratings_df.rating).alias(\"average\"))\nprint 'movie_ids_with_avg_ratings_df:'\nmovie_ids_with_avg_ratings_df.show(3, truncate=False)\n\n# Note: movie_names_df is a temporary variable, used only to separate the steps necessary\n# to create the movie_names_with_avg_ratings_df DataFrame.\nmovies_df_good=movies_df.withColumnRenamed(\"ID\",\"movieId\")\nmovie_names_df = movie_ids_with_avg_ratings_df.join(movies_df_good,\"movieId\").select(\"average\", \"title\", \"count\", \"movieId\")\n\nmovie_names_with_avg_ratings_df = movie_names_df\nprint 'movie_names_with_avg_ratings_df:'\nmovie_names_with_avg_ratings_df.show(3, truncate=False)"],"metadata":{},"outputs":[],"execution_count":20},{"cell_type":"code","source":["# TEST Movies with Highest Average Ratings (1a)\nTest.assertEquals(movie_ids_with_avg_ratings_df.count(), 26744,\n                'incorrect movie_ids_with_avg_ratings_df.count() (expected 26744)')\nmovie_ids_with_ratings_take_ordered = movie_ids_with_avg_ratings_df.orderBy('MovieID').take(3)\n_take_0 = movie_ids_with_ratings_take_ordered[0]\n_take_1 = movie_ids_with_ratings_take_ordered[1]\n_take_2 = movie_ids_with_ratings_take_ordered[2]\nTest.assertTrue(_take_0[0] == 1 and _take_0[1] == 49695,\n                'incorrect count of ratings for movie with ID {0} (expected 49695)'.format(_take_0[0]))\nTest.assertEquals(round(_take_0[2], 2), 3.92, \"Incorrect average for movie ID {0}. Expected 3.92\".format(_take_0[0]))\n\nTest.assertTrue(_take_1[0] == 2 and _take_1[1] == 22243,\n                'incorrect count of ratings for movie with ID {0} (expected 22243)'.format(_take_1[0]))\nTest.assertEquals(round(_take_1[2], 2), 3.21, \"Incorrect average for movie ID {0}. Expected 3.21\".format(_take_1[0]))\n\nTest.assertTrue(_take_2[0] == 3 and _take_2[1] == 12735,\n                'incorrect count of ratings for movie with ID {0} (expected 12735)'.format(_take_2[0]))\nTest.assertEquals(round(_take_2[2], 2), 3.15, \"Incorrect average for movie ID {0}. Expected 3.15\".format(_take_2[0]))\n\n\nTest.assertEquals(movie_names_with_avg_ratings_df.count(), 26744,\n                  'incorrect movie_names_with_avg_ratings_df.count() (expected 26744)')\nmovie_names_with_ratings_take_ordered = movie_names_with_avg_ratings_df.orderBy(['average', 'title']).take(3)\nresult = [(r['average'], r['title'], r['count'], r['movieId']) for r in movie_names_with_ratings_take_ordered]\nTest.assertEquals(result,\n                  [(0.5, u'13 Fighting Men (1960)', 1, 109355),\n                   (0.5, u'20 Years After (2008)', 1, 131062),\n                   (0.5, u'3 Holiday Tails (Golden Christmas 2: The Second Tail, A) (2011)', 1, 111040)],\n                  'incorrect top 3 entries in movie_names_with_avg_ratings_df')"],"metadata":{},"outputs":[],"execution_count":21},{"cell_type":"markdown","source":["### (1b) Movies with Highest Average Ratings and at least 500 reviews\n\nNow that we have a DataFrame of the movies with highest average ratings, we can use Spark to determine the 20 movies with highest average ratings and at least 500 reviews.\n\nAdd a single DataFrame transformation (in place of `<FILL_IN>`, below) to limit the results to movies with ratings from at least 500 people."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL IN> with appropriate code\nmovies_with_500_ratings_or_more = movie_names_with_avg_ratings_df.filter(movie_names_with_avg_ratings_df[2]>=500).orderBy(movie_names_with_avg_ratings_df.average.desc())\nprint 'Movies with highest ratings:'\nmovies_with_500_ratings_or_more.show(20, truncate=False)"],"metadata":{},"outputs":[],"execution_count":23},{"cell_type":"code","source":["# TEST Movies with Highest Average Ratings and at least 500 Reviews (1b)\n\nTest.assertEquals(movies_with_500_ratings_or_more.count(), 4489,\n                  'incorrect movies_with_500_ratings_or_more.count(). Expected 4489.')\ntop_20_results = [(r['average'], r['title'], r['count']) for r in movies_with_500_ratings_or_more.orderBy(F.desc('average')).take(20)]\n\nTest.assertEquals(top_20_results,\n                  [(4.446990499637029, u'Shawshank Redemption, The (1994)', 63366),\n                   (4.364732196832306, u'Godfather, The (1972)', 41355),\n                   (4.334372207803259, u'Usual Suspects, The (1995)', 47006),\n                   (4.310175010988133, u\"Schindler's List (1993)\", 50054),\n                   (4.275640557704942, u'Godfather: Part II, The (1974)', 27398),\n                   (4.2741796572216, u'Seven Samurai (Shichinin no samurai) (1954)', 11611),\n                   (4.271333600779414, u'Rear Window (1954)', 17449),\n                   (4.263182346109176, u'Band of Brothers (2001)', 4305),\n                   (4.258326830670664, u'Casablanca (1942)', 24349),\n                   (4.256934865900383, u'Sunset Blvd. (a.k.a. Sunset Boulevard) (1950)', 6525),\n                   (4.24807897901911, u\"One Flew Over the Cuckoo's Nest (1975)\", 29932),\n                   (4.247286821705426, u'Dr. Strangelove or: How I Learned to Stop Worrying and Love the Bomb (1964)', 23220),\n                   (4.246001523229246, u'Third Man, The (1949)', 6565),\n                   (4.235410064157069, u'City of God (Cidade de Deus) (2002)', 12937),\n                   (4.2347902097902095, u'Lives of Others, The (Das leben der Anderen) (2006)', 5720),\n                   (4.233538107122288, u'North by Northwest (1959)', 15627),\n                   (4.2326233183856505, u'Paths of Glory (1957)', 3568),\n                   (4.227123123722136, u'Fight Club (1999)', 40106),\n                   (4.224281931146873, u'Double Indemnity (1944)', 4909),\n                   (4.224137931034483, u'12 Angry Men (1957)', 12934)],\n                  'Incorrect top 20 movies with 500 or more ratings')"],"metadata":{},"outputs":[],"execution_count":24},{"cell_type":"markdown","source":["Using a threshold on the number of reviews is one way to improve the recommendations, but there are many other good ways to improve quality. For example, you could weight ratings by the number of ratings."],"metadata":{}},{"cell_type":"markdown","source":["## Part 2: Collaborative Filtering\nIn this course, you have learned about many of the basic transformations and actions that Spark allows us to apply to distributed datasets.  Spark also exposes some higher level functionality; in particular, Machine Learning using a component of Spark called [MLlib][mllib].  In this part, you will learn how to use MLlib to make personalized movie recommendations using the movie data we have been analyzing.\n\n<img src=\"https://courses.edx.org/c4x/BerkeleyX/CS100.1x/asset/Collaborative_filtering.gif\" alt=\"collaborative filtering\" style=\"float: right\"/>\n\nWe are going to use a technique called [collaborative filtering][collab]. Collaborative filtering is a method of making automatic predictions (filtering) about the interests of a user by collecting preferences or taste information from many users (collaborating). The underlying assumption of the collaborative filtering approach is that if a person A has the same opinion as a person 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 person chosen randomly. You can read more about collaborative filtering [here][collab2].\n\nThe image at the right (from [Wikipedia][collab]) shows an example of predicting of the user's rating using collaborative filtering. At first, people rate different items (like videos, images, games). After that, the system is making predictions about a user's rating for an item, which the user has not rated yet. These predictions are built upon the existing ratings of other users, who have similar ratings with the active user. For instance, in the image below the system has made a prediction, that the active user will not like the video.\n\n<br clear=\"all\"/>\n\n----\n\nFor movie recommendations, we start with a matrix whose entries are movie ratings by users (shown in red in the diagram below).  Each column represents a user (shown in green) and each row represents a particular movie (shown in blue).\n\nSince not all users have rated all movies, we do not know all of the entries in this matrix, which is precisely why we need collaborative filtering.  For each user, we have ratings for only a subset of the movies.  With collaborative filtering, the idea is to approximate the ratings matrix by factorizing it as the product of two matrices: one that describes properties of each user (shown in green), and one that describes properties of each movie (shown in blue).\n\n<img alt=\"factorization\" src=\"http://spark-mooc.github.io/web-assets/images/matrix_factorization.png\" style=\"width: 885px\"/>\n<br clear=\"all\"/>\n\nWe want to select these two matrices such that the error for the users/movie pairs where we know the correct ratings is minimized.  The [Alternating Least Squares][als] algorithm does this by first randomly filling the users matrix with values and then optimizing the value of the movies such that the error is minimized.  Then, it holds the movies matrix constrant and optimizes the value of the user's matrix.  This alternation between which matrix to optimize is the reason for the \"alternating\" in the name.\n\nThis optimization is what's being shown on the right in the image above.  Given a fixed set of user factors (i.e., values in the users matrix), we use the known ratings to find the best values for the movie factors using the optimization written at the bottom of the figure.  Then we \"alternate\" and pick the best user factors given fixed movie factors.\n\nFor a simple example of what the users and movies matrices might look like, check out the [videos from Lecture 2][videos] or the [slides from Lecture 8][slides]\n[videos]: https://courses.edx.org/courses/course-v1:BerkeleyX+CS110x+2T2016/courseware/9d251397874d4f0b947b606c81ccf83c/3cf61a8718fe4ad5afcd8fb35ceabb6e/\n[slides]: https://d37djvu3ytnwxt.cloudfront.net/assets/courseware/v1/fb269ff9a53b669a46d59e154b876d78/asset-v1:BerkeleyX+CS110x+2T2016+type@asset+block/Lecture2s.pdf\n[als]: https://en.wikiversity.org/wiki/Least-Squares_Method\n[mllib]: http://spark.apache.org/docs/1.6.2/mllib-guide.html\n[collab]: https://en.wikipedia.org/?title=Collaborative_filtering\n[collab2]: http://recommender-systems.org/collaborative-filtering/"],"metadata":{}},{"cell_type":"markdown","source":["### (2a) Creating a Training Set\n\nBefore we jump into using machine learning, we need to break up the `ratings_df` dataset into three pieces:\n* A training set (DataFrame), which we will use to train models\n* A validation set (DataFrame), which we will use to choose the best model\n* A test set (DataFrame), which we will use for our experiments\n\nTo randomly split the dataset into the multiple groups, we can use the pySpark [randomSplit()](http://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html#pyspark.sql.DataFrame.randomSplit) transformation. `randomSplit()` takes a set of splits and a seed and returns multiple DataFrames."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL_IN> with the appropriate code.\n\n# We'll hold out 60% for training, 20% of our data for validation, and leave 20% for testing\nseed = 1800009193L\n(split_60_df, split_a_20_df, split_b_20_df) = ratings_df.randomSplit((.60,.20,.20),seed)\n\n# Let's cache these datasets for performance\ntraining_df = split_60_df.cache()\nvalidation_df = split_a_20_df.cache()\ntest_df = split_b_20_df.cache()\n\nprint('Training: {0}, validation: {1}, test: {2}\\n'.format(\n  training_df.count(), validation_df.count(), test_df.count())\n)\ntraining_df.show(3)\nvalidation_df.show(3)\ntest_df.show(3)"],"metadata":{},"outputs":[],"execution_count":28},{"cell_type":"code","source":["# TEST Creating a Training Set (2a)\nTest.assertEquals(training_df.count(), 12001389, \"Incorrect training_df count. Expected 12001389\")\nTest.assertEquals(validation_df.count(), 4003694, \"Incorrect validation_df count. Expected 4003694\")\nTest.assertEquals(test_df.count(), 3995180, \"Incorrect test_df count. Expected 3995180\")\n\nTest.assertEquals(training_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 5952) & (ratings_df.rating == 5.0)).count(), 1)\nTest.assertEquals(training_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 1193) & (ratings_df.rating == 3.5)).count(), 1)\nTest.assertEquals(training_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 1196) & (ratings_df.rating == 4.5)).count(), 1)\n\nTest.assertEquals(validation_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 296) & (ratings_df.rating == 4.0)).count(), 1)\nTest.assertEquals(validation_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 32) & (ratings_df.rating == 3.5)).count(), 1)\nTest.assertEquals(validation_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 6888) & (ratings_df.rating == 3.0)).count(), 1)\n\nTest.assertEquals(test_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 4993) & (ratings_df.rating == 5.0)).count(), 1)\nTest.assertEquals(test_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 4128) & (ratings_df.rating == 4.0)).count(), 1)\nTest.assertEquals(test_df.filter((ratings_df.userId == 1) & (ratings_df.movieId == 4915) & (ratings_df.rating == 3.0)).count(), 1)"],"metadata":{},"outputs":[],"execution_count":29},{"cell_type":"markdown","source":["After splitting the dataset, your training set has about 12 million entries and the validation and test sets each have about 4 million entries. (The exact number of entries in each dataset varies slightly due to the random nature of the `randomSplit()` transformation.)"],"metadata":{}},{"cell_type":"markdown","source":["### (2b) Alternating Least Squares\n\nIn this part, we will use the Apache Spark ML Pipeline implementation of Alternating Least Squares, [ALS](http://spark.apache.org/docs/1.6.2/api/python/pyspark.ml.html#pyspark.ml.recommendation.ALS). ALS takes a training dataset (DataFrame) and several parameters that control the model creation process. To determine the best values for the parameters, we will use ALS to train several models, and then we will select the best model and use the parameters from that model in the rest of this lab exercise.\n\nThe process we will use for determining the best model is as follows:\n1. Pick a set of model parameters. The most important parameter to model is the *rank*, which is the number of columns in the Users matrix (green in the diagram above) or the number of rows in the Movies matrix (blue in the diagram above). In general, a lower rank will mean higher error on the training dataset, but a high rank may lead to [overfitting](https://en.wikipedia.org/wiki/Overfitting).  We will train models with ranks of 4, 8, and 12 using the `training_df` dataset.\n\n2. Set the appropriate parameters on the `ALS` object:\n    * The \"User\" column will be set to the values in our `userId` DataFrame column.\n    * The \"Item\" column will be set to the values in our `movieId` DataFrame column.\n    * The \"Rating\" column will be set to the values in our `rating` DataFrame column.\n    * We'll using a regularization parameter of 0.1.\n\n   **Note**: Read the documentation for the [ALS](http://spark.apache.org/docs/1.6.2/api/python/pyspark.ml.html#pyspark.ml.recommendation.ALS) class **carefully**. It will help you accomplish this step.\n3. Have the ALS output transformation (i.e., the result of [ALS.fit()](http://spark.apache.org/docs/1.6.2/api/python/pyspark.ml.html#pyspark.ml.recommendation.ALS.fit)) produce a _new_ column\n   called \"prediction\" that contains the predicted value.\n\n4. Create multiple models using [ALS.fit()](http://spark.apache.org/docs/1.6.2/api/python/pyspark.ml.html#pyspark.ml.recommendation.ALS.fit), one for each of our rank values. We'll fit\n   against the training data set (`training_df`).\n\n5. For each model, we'll run a prediction against our validation data set (`validation_df`) and check the error.\n\n6. We'll keep the model with the best error rate.\n\n#### Why are we doing our own cross-validation?\n\nA challenge for collaborative filtering is how to provide ratings to a new user (a user who has not provided *any* ratings at all). Some recommendation systems choose to provide new users with a set of default ratings (e.g., an average value across all ratings), while others choose to provide no ratings for new users. Spark's ALS algorithm yields a NaN (`Not a Number`) value when asked to provide a rating for a new user.\n\nUsing the ML Pipeline's [CrossValidator](http://spark.apache.org/docs/1.6.2/api/python/pyspark.ml.html#pyspark.ml.tuning.CrossValidator) with ALS is thus problematic, because cross validation involves dividing the training data into a set of folds (e.g., three sets) and then using those folds for testing and evaluating the parameters during the parameter grid search process. It is likely that some of the folds will contain users that are not in the other folds, and, as a result, ALS produces NaN values for those new users. When the CrossValidator uses the Evaluator (RMSE) to compute an error metric, the RMSE algorithm will return NaN. This will make *all* of the parameters in the parameter grid appear to be equally good (or bad).\n\nYou can read the discussion on [Spark JIRA 14489](https://issues.apache.org/jira/browse/SPARK-14489) about this issue. There are proposed workarounds of having ALS provide default values or having RMSE drop NaN values. Both introduce potential issues. We have chosen to have RMSE drop NaN values. While this does not solve the underlying issue of ALS not predicting a value for a new user, it does provide some evaluation value. We manually implement the parameter grid search process using a for loop (below) and remove the NaN values before using RMSE.\n\nFor a production application, you would want to consider the tradeoffs in how to handle new users.\n\n**Note**: This cell will likely take a couple of minutes to run."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL IN> with appropriate code\n# This step is broken in ML Pipelines: https://issues.apache.org/jira/browse/SPARK-14489\nfrom pyspark.ml.recommendation import ALS\n\n# Let's initialize our ALS learner\nals = ALS()\n\n# Now we set the parameters for the method\nals.setMaxIter(5)\\\n   .setSeed(seed)\\\n   .setRegParam(0.1)\\\n   .setItemCol('movieId')\\\n   .setUserCol('userId')\\\n   .setRatingCol('rating')\n\n# Now let's compute an evaluation metric for our test dataset\nfrom pyspark.ml.evaluation import RegressionEvaluator\n\n# Create an RMSE evaluator using the label and predicted columns\nreg_eval = RegressionEvaluator(predictionCol=\"prediction\", labelCol=\"rating\", metricName=\"rmse\")\n\ntolerance = 0.03\nranks = [4, 8, 12]\nerrors = [0, 0, 0]\nmodels = [0, 0, 0]\nerr = 0\nmin_error = float('inf')\nbest_rank = -1\nfor rank in ranks:\n  # Set the rank here:\n  als.setRank(rank)\n  # Create the model with these parameters.\n  model = als.fit(training_df)\n  # Run the model to create a prediction. Predict against the validation_df.\n  predict_df = model.transform(validation_df)\n\n  # Remove NaN values from prediction (due to SPARK-14489)\n  predicted_ratings_df = predict_df.filter(predict_df.prediction != float('nan'))\n\n  # Run the previously created RMSE evaluator, reg_eval, on the predicted_ratings_df DataFrame\n  error = reg_eval.evaluate(predicted_ratings_df)\n  errors[err] = error\n  models[err] = model\n  print 'For rank %s the RMSE is %s' % (rank, error)\n  if error < min_error:\n    min_error = error\n    best_rank = err\n  err += 1\n\nals.setRank(ranks[best_rank])\nprint 'The best model was trained with rank %s' % ranks[best_rank]\nmy_model = models[best_rank]"],"metadata":{},"outputs":[],"execution_count":32},{"cell_type":"code","source":["# TEST\nTest.assertEquals(round(min_error, 2), 0.81, \"Unexpected value for best RMSE. Expected rounded value to be 0.81. Got {0}\".format(round(min_error, 2)))\nTest.assertEquals(ranks[best_rank], 12, \"Unexpected value for best rank. Expected 12. Got {0}\".format(ranks[best_rank]))\nTest.assertEqualsHashed(als.getItemCol(), \"18f0e2357f8829fe809b2d95bc1753000dd925a6\", \"Incorrect choice of {0} for ALS item column.\".format(als.getItemCol()))\nTest.assertEqualsHashed(als.getUserCol(), \"db36668fa9a19fde5c9676518f9e86c17cabf65a\", \"Incorrect choice of {0} for ALS user column.\".format(als.getUserCol()))\nTest.assertEqualsHashed(als.getRatingCol(), \"3c2d687ef032e625aa4a2b1cfca9751d2080322c\", \"Incorrect choice of {0} for ALS rating column.\".format(als.getRatingCol()))"],"metadata":{},"outputs":[],"execution_count":33},{"cell_type":"markdown","source":["### (2c) Testing Your Model\n\nSo far, we used the `training_df` and `validation_df` datasets to select the best model.  Since we used these two datasets to determine what model is best, we cannot use them to test how good the model is; otherwise, we would be very vulnerable to [overfitting](https://en.wikipedia.org/wiki/Overfitting).  To decide how good our model is, we need to use the `test_df` dataset.  We will use the `best_rank` you determined in part (2b) to create a model for predicting the ratings for the test dataset and then we will compute the RMSE.\n\nThe steps you should perform are:\n* Run a prediction, using `my_model` as created above, on the test dataset (`test_df`), producing a new `predict_df` DataFrame.\n* Filter out unwanted NaN values (necessary because of [a bug in Spark](https://issues.apache.org/jira/browse/SPARK-14489)). We've supplied this piece of code for you.\n* Use the previously created RMSE evaluator, `reg_eval` to evaluate the filtered DataFrame."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL_IN> with the appropriate code\n# In ML Pipelines, this next step has a bug that produces unwanted NaN values. We\n# have to filter them out. See https://issues.apache.org/jira/browse/SPARK-14489\npredict_df = my_model.transform(test_df)\n\n# Remove NaN values from prediction (due to SPARK-14489)\npredicted_test_df = predict_df.filter(predict_df.prediction != float('nan'))\n\n# Run the previously created RMSE evaluator, reg_eval, on the predicted_test_df DataFrame\ntest_RMSE = reg_eval.evaluate(predicted_test_df)\n\nprint('The model had a RMSE on the test set of {0}'.format(test_RMSE))"],"metadata":{},"outputs":[],"execution_count":35},{"cell_type":"code","source":["# TEST Testing Your Model (2c)\nTest.assertTrue(abs(test_RMSE - 0.809624038485) < tolerance, 'incorrect test_RMSE: {0:.11f}'.format(test_RMSE))"],"metadata":{},"outputs":[],"execution_count":36},{"cell_type":"markdown","source":["### (2d) Comparing Your Model\n\nLooking at the RMSE for the results predicted by the model versus the values in the test set is one way to evalute the quality of our model. Another way to evaluate the model is to evaluate the error from a test set where every rating is the average rating for the training set.\n\nThe steps you should perform are:\n* Use the `training_df` to compute the average rating across all movies in that training dataset.\n* Use the average rating that you just determined and the `test_df` to create a DataFrame (`test_for_avg_df`) with a `prediction` column containing the average rating. **HINT**: You'll want to use the `lit()` function,\n  from `pyspark.sql.functions`, available here as `F.lit()`.\n* Use our previously created `reg_eval` object to evaluate the `test_for_avg_df` and calculate the RMSE."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL_IN> with the appropriate code.\n# Compute the average rating\navg_rating_df = training_df.agg(F.avg(training_df.rating)).alias('average')\n\n# Extract the average rating value. (This is row 0, column 0.)\ntraining_avg_rating = avg_rating_df.collect()[0][0]\n\nprint('The average rating for movies in the training set is {0}'.format(training_avg_rating))\n\n# Add a column with the average rating\ntest_for_avg_df = training_df.withColumn('prediction', F.lit(training_avg_rating))\n\n# Run the previously created RMSE evaluator, reg_eval, on the test_for_avg_df DataFrame\ntest_avg_RMSE = reg_eval.evaluate(test_for_avg_df)\n\nprint(\"The RMSE on the average set is {0}\".format(test_avg_RMSE))"],"metadata":{},"outputs":[],"execution_count":38},{"cell_type":"code","source":["# TEST Comparing Your Model (2d)\nTest.assertTrue(abs(training_avg_rating - 3.52547984237) < 0.000001,\n                'incorrect training_avg_rating (expected 3.52547984237): {0:.11f}'.format(training_avg_rating))\nTest.assertTrue(abs(test_avg_RMSE - 1.0519743756) < 0.000001,\n                'incorrect test_avg_RMSE (expected 1.0519743756): {0:.11f}'.format(test_avg_RMSE))"],"metadata":{},"outputs":[],"execution_count":39},{"cell_type":"markdown","source":["You now have code to predict how users will rate movies!"],"metadata":{}},{"cell_type":"markdown","source":["## Part 3: Predictions for Yourself\nThe ultimate goal of this lab exercise is to predict what movies to recommend to yourself.  In order to do that, you will first need to add ratings for yourself to the `ratings_df` dataset."],"metadata":{}},{"cell_type":"markdown","source":["**(3a) Your Movie Ratings**\n\nTo help you provide ratings for yourself, we have included the following code to list the names and movie IDs of the 50 highest-rated movies from `movies_with_500_ratings_or_more` which we created in part 1 the lab."],"metadata":{}},{"cell_type":"code","source":["print 'Most rated movies:'\nprint '(average rating, movie name, number of reviews, movie ID)'\ndisplay(movies_with_500_ratings_or_more.orderBy(movies_with_500_ratings_or_more['average'].desc()).take(50))"],"metadata":{},"outputs":[],"execution_count":43},{"cell_type":"markdown","source":["The user ID 0 is unassigned, so we will use it for your ratings. We set the variable `my_user_ID` to 0 for you. Next, create a new DataFrame called `my_ratings_df`, with your ratings for at least 10 movie ratings. Each entry should be formatted as `(my_user_id, movieID, rating)`.  As in the original dataset, ratings should be between 1 and 5 (inclusive). If you have not seen at least 10 of these movies, you can increase the parameter passed to `take()` in the above cell until there are 10 movies that you have seen (or you can also guess what your rating would be for movies you have not seen)."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL IN> with appropriate code\nfrom pyspark.sql import Row\nmy_user_id = 0\n\n# Note that the movie IDs are the *last* number on each line. A common error was to use the number of ratings as the movie ID.\nmy_rated_movies = [\n     (my_user_id, 858, 5),(my_user_id, 50, 4), (my_user_id, 527, 2), (my_user_id, 318, 5), (my_user_id, 904, 4), (my_user_id,6016,5 ), (my_user_id, 5618, 5), (my_user_id, 1213, 5), (my_user_id, 4226, 3), (my_user_id, 296, 5), (my_user_id, 2571, 4)\n     # The format of each line is (my_user_id, movie ID, your rating)\n     # For example, to give the movie \"Star Wars: Episode IV - A New Hope (1977)\" a five rating, you would add the following line:\n     #   (my_user_id, 260, 5),\n]\n\nmy_ratings_df = sqlContext.createDataFrame(my_rated_movies, ['userId','movieId','rating'])\nprint 'My movie ratings:'\ndisplay(my_ratings_df.limit(10))"],"metadata":{},"outputs":[],"execution_count":45},{"cell_type":"markdown","source":["### (3b) Add Your Movies to Training Dataset\n\nNow that you have ratings for yourself, you need to add your ratings to the `training` dataset so that the model you train will incorporate your preferences.  Spark's [unionAll()](http://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html#pyspark.sql.DataFrame.unionAll) transformation combines two DataFrames; use `unionAll()` to create a new training dataset that includes your ratings and the data in the original training dataset."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL IN> with appropriate code\ntraining_with_my_ratings_df = my_ratings_df.unionAll(training_df)\n\nprint ('The training dataset now has %s more entries than the original training dataset' %\n       (training_with_my_ratings_df.count() - training_df.count()))\nassert (training_with_my_ratings_df.count() - training_df.count()) == my_ratings_df.count()"],"metadata":{},"outputs":[],"execution_count":47},{"cell_type":"markdown","source":["### (3c) Train a Model with Your Ratings\n\nNow, train a model with your ratings added and the parameters you used in in part (2b) and (2c). Mke sure you include **all** of the parameters.\n\n**Note**: This cell will take about 30 seconds to run."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL IN> with appropriate code\n\n# Reset the parameters for the ALS object.\nals.setPredictionCol(\"prediction\")\\\n   .setMaxIter(5)\\\n   .setSeed(seed)\\\n   .setItemCol('movieId')\\\n   .setUserCol('userId')\\\n   .setRatingCol('rating')\n\n# Create the model with these parameters.\nmy_ratings_model = als.fit(training_with_my_ratings_df)"],"metadata":{},"outputs":[],"execution_count":49},{"cell_type":"markdown","source":["### (3d) Check RMSE for the New Model with Your Ratings\n\nCompute the RMSE for this new model on the test set.\n* Run your model (the one you just trained) against the test data set in `test_df`.\n* Then, use our previously-computed `reg_eval` object to compute the RMSE of your ratings."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL IN> with appropriate code\nmy_predict_df = my_ratings_model.transform(test_df)\n\n# Remove NaN values from prediction (due to SPARK-14489)\npredicted_test_my_ratings_df = my_predict_df.filter(my_predict_df.prediction != float('nan'))\n\n# Run the previously created RMSE evaluator, reg_eval, on the predicted_test_my_ratings_df DataFrame\ntest_RMSE_my_ratings = reg_eval.evaluate(predicted_test_my_ratings_df)\nprint('The model had a RMSE on the test set of {0}'.format(test_RMSE_my_ratings))"],"metadata":{},"outputs":[],"execution_count":51},{"cell_type":"markdown","source":["### (3e) Predict Your Ratings\n\nSo far, we have only computed the error of the model.  Next, let's predict what ratings you would give to the movies that you did not already provide ratings for.\n\nThe steps you should perform are:\n* Filter out the movies you already rated manually. (Use the `my_rated_movie_ids` variable.) Put the results in a new `not_rated_df`.\n\n   **Hint**: The [Column.isin()](http://spark.apache.org/docs/1.6.2/api/python/pyspark.sql.html#pyspark.sql.Column.isin)\n   method, as well as the `~` (\"not\") DataFrame logical operator, may come in handy here. Here's an example of using `isin()`:\n\n```\n    > df1 = sqlContext.createDataFrame([(\"Jim\", 10), (\"Julie\", 9), (\"Abdul\", 20), (\"Mireille\", 19)], [\"name\", \"age\"])\n    > df1.show()\n    +--------+---+\n    |    name|age|\n    +--------+---+\n    |     Jim| 10|\n    |   Julie|  9|\n    |   Abdul| 20|\n    |Mireille| 19|\n    +--------+---+\n\n    > names_to_delete = (\"Julie\", \"Abdul\") # this is just a Python tuple\n    > df2 = df1.filter(~ df1[\"name\"].isin(names_to_delete)) # \"NOT IN\"\n    > df2.show()\n    +--------+---+\n    |    name|age|\n    +--------+---+\n    |     Jim| 10|\n    |   Julie|  9|\n    +--------+---+\n```\n\n* Transform `not_rated_df` into `my_unrated_movies_df` by:\n    - renaming the \"ID\" column to \"movieId\"\n    - adding a \"userId\" column with the value contained in the `my_user_id` variable defined above.\n\n* Create a `predicted_ratings_df` DataFrame by applying `my_ratings_model` to `my_unrated_movies_df`."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL_IN> with the appropriate code\n\n# Create a list of my rated movie IDs\nmy_rated_movie_ids = [x[1] for x in my_rated_movies]\n\n# Filter out the movies I already rated.\nnot_rated_df = movies_df.filter(~movies_df['ID'].isin(my_rated_movie_ids))\n\n\n# Rename the \"ID\" column to be \"movieId\", and add a column with my_user_id as \"userId\".\nmy_unrated_movies_df = not_rated_df.selectExpr(\"ID as movieId\").withColumn('userId', F.lit(my_user_id))\n# Use my_rating_model to predict ratings for the movies that I did not manually rate.\nraw_predicted_ratings_df = my_ratings_model.transform(my_unrated_movies_df)\n\npredicted_ratings_df = raw_predicted_ratings_df.filter(raw_predicted_ratings_df['prediction'] != float('nan'))"],"metadata":{},"outputs":[],"execution_count":53},{"cell_type":"markdown","source":["### (3f) Predict Your Ratings\n\nWe have our predicted ratings. Now we can print out the 25 movies with the highest predicted ratings.\n\nThe steps you should perform are:\n* Join your `predicted_ratings_df` DataFrame with the `movie_names_with_avg_ratings_df` DataFrame to obtain the ratings counts for each movie.\n* Sort the resulting DataFrame (`predicted_with_counts_df`) by predicted rating (highest ratings first), and remove any ratings with a count of 75 or less.\n* Print the top 25 movies that remain."],"metadata":{}},{"cell_type":"code","source":["# TODO: Replace <FILL_IN> with the appropriate code\n\npredicted_with_counts_df = predicted_ratings_df.join(movie_names_with_avg_ratings_df,\"movieId\").select(\"prediction\",\"average\", \"title\", \"count\", \"movieId\")\npredicted_highest_rated_movies_df = predicted_with_counts_df.orderBy(predicted_with_counts_df.prediction.desc()).filter(predicted_with_counts_df[3]>75)\n\nprint ('My 25 highest rated movies as predicted (for movies with more than 75 reviews):')\ndisplay(predicted_highest_rated_movies_df)"],"metadata":{},"outputs":[],"execution_count":55},{"cell_type":"markdown","source":["## Appendix A: Submitting Your Exercises to the Autograder\n\nThis section guides you through Step 2 of the grading process (\"Submit to Autograder\").\n\nOnce you confirm that your lab notebook is passing all tests, you can submit it first to the course autograder and then second to the edX website to receive a grade.\n\n** Note that you can only submit to the course autograder once every 1 minute. **"],"metadata":{}},{"cell_type":"markdown","source":["### Step 2(a): Restart your cluster by clicking on the dropdown next to your cluster name and selecting \"Restart Cluster\".\n\nYou can do this step in either notebook, since there is one cluster for your notebooks.\n\n<img src=\"http://spark-mooc.github.io/web-assets/images/submit_restart.png\" alt=\"Drawing\" />"],"metadata":{}},{"cell_type":"markdown","source":["### Step 2(b): _IN THIS NOTEBOOK_, click on \"Run All\" to run all of the cells.\n\n<img src=\"http://spark-mooc.github.io/web-assets/images/submit_runall.png\" alt=\"Drawing\" style=\"height: 80px\"/>\n\nThis step will take some time.\n\nWait for your cluster to finish running the cells in your lab notebook before proceeding."],"metadata":{}},{"cell_type":"markdown","source":["### Step 2(c): Publish this notebook\n\nPublish _this_ notebook by clicking on the \"Publish\" button at the top.\n\n<img src=\"http://spark-mooc.github.io/web-assets/images/Lab0_Publish0.png\" alt=\"Drawing\" style=\"height: 150px\"/>\n\nWhen you click on the button, you will see the following popup.\n\n<img src=\"http://spark-mooc.github.io/web-assets/images/Lab0_Publish1.png\" alt=\"Drawing\" />\n\nWhen you click on \"Publish\", you will see a popup with your notebook's public link. **Copy the link and set the `notebook_URL` variable in the AUTOGRADER notebook (not this notebook).**\n\n<img src=\"http://spark-mooc.github.io/web-assets/images/Lab0_Publish2.png\" alt=\"Drawing\" />"],"metadata":{}},{"cell_type":"markdown","source":["### Step 2(d): Set the notebook URL and Lab ID in the Autograder notebook, and run it\n\nGo to the Autograder notebook and paste the link you just copied into it, so that it is assigned to the `notebook_url` variable.\n\n```\nnotebook_url = \"...\" # put your URL here\n```\n\nThen, find the line that looks like this:\n\n```\nlab = <FILL IN>\n```\nand change `<FILL IN>` to \"CS110x-lab2\":\n\n```\nlab = \"CS110x-lab2\"\n```\n\nThen, run the Autograder notebook to submit your lab."],"metadata":{}},{"cell_type":"markdown","source":["### <img src=\"http://spark-mooc.github.io/web-assets/images/oops.png\" style=\"height: 200px\"/> If things go wrong\n\nIt's possible that your notebook looks fine to you, but fails in the autograder. (This can happen when you run cells out of order, as you're working on your notebook.) If that happens, just try again, starting at the top of Appendix A."],"metadata":{}}],"metadata":{"name":"cs110_lab2_als_prediction","notebookId":498761534254159},"nbformat":4,"nbformat_minor":0}