How do you build a movie recommendation system via SVD using Apache Spark?

Getting and processing the data 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. The list of task we can pre-compute includes: Loading and parsing the dataset. Persisting the resulting RDD for later use. Building the recommender model using the complete dataset. Persist the dataset for later use. This notebook explains the first of these tasks. complete_dataset_url = 'http://files.grouplens.org/datasets/movielens/ml-latest.zip' small_dataset_url = 'http://files.grouplens.org/datasets/movielens/ml-latest-small.zip' We also need to define download locations. import os datasets_path = os.path.join('..', 'datasets') complete_dataset_path = os.path.join(datasets_path, 'ml-latest.zip') small_dataset_path = os.path.join(datasets_path, 'ml-latest-small.zip') import urllib small_f = urllib.urlretrieve (small_dataset_url, small_dataset_path) complete_f = urllib.urlretrieve (complete_dataset_url, complete_dataset_path) 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. import zipfile with zipfile.ZipFile(small_dataset_path, "r") as z: z.extractall(datasets_path) with zipfile.ZipFile(complete_dataset_path, "r") as z: z.extractall(datasets_path) Loading and parsing datasets Now we are ready to read in each of the files and create an RDD consisting of parsed lines. Each line in the ratings dataset (ratings.csv) is formatted as: userId,movieId,rating,timestamp Each line in the movies (movies.csv) dataset is formatted as: movieId,title,genres Were genres has the format: Genre1|Genre2|Genre3... The tags file (tags.csv) has the format: userId,movieId,tag,timestamp And finally, the links.csv file has the format: movieId,imdbId,tmdbId The format of these files is uniform and simple, so we can use Python split() to parse their lines once they are loaded into RDDs. Parsing the movies and ratings files yields two RDDs: 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. 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. So let's load the raw ratings data. We need to filter out the header, included in each file. small_ratings_file = os.path.join(datasets_path, 'ml-latest-small', 'ratings.csv') small_ratings_raw_data = sc.textFile(small_ratings_file) small_ratings_raw_data_header = small_ratings_raw_data.take(1)[0] Now we can parse the raw data into a new RDD small_ratings_data = small_ratings_raw_data.filter(lambda line: line!=small_ratings_raw_data_header)\ .map(lambda line: line.split(",")).map(lambda tokens: (tokens[0],tokens[1],tokens[2])).cache() 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. small_ratings_data.take(3) [(u'1', u'6', u'2.0'), (u'1', u'22', u'3.0'), (u'1', u'32', u'2.0')] We proceed in a similar way with the movies.csv file. small_movies_file = os.path.join(datasets_path, 'ml-latest-small', 'movies.csv') small_movies_raw_data = sc.textFile(small_movies_file) small_movies_raw_data_header = small_movies_raw_data.take(1)[0] small_movies_data = small_movies_raw_data.filter(lambda line: line!=small_movies_raw_data_header)\ .map(lambda line: line.split(",")).map(lambda tokens: (tokens[0],tokens[1])).cache() small_movies_data.take(3) [(u'1', u'Toy Story (1995)'), (u'2', u'Jumanji (1995)'), (u'3', u'Grumpier Old Men (1995)')] read less