<jupyter_start><jupyter_text>Movie recommender system with Spark machine learningIn this Jupyter notebook, you will use Spark and the Spark machine learning library to build a recommender system for movies with a data set from MovieLens. You'll also learn how to manage Spark performance.Some familiarity with Python is recommended. This notebook runs on Python 3.6 with Spark. MovieLensMovieLens is a project developed by GroupLens, a research laboratory at the University of Minnesota. MovieLens provides an online movie recommender application that uses anonymously-collected data to improve recommender algorithms. Anyone can try the app for free and get movies recommendations. To help people develop the best recommendation algorithms, MovieLens also released several data sets. In this notebook, you'll use the latest data set, which has two sizes.The full data set consists of more than 24 million ratings across more than 40,000 movies by more than 250,000 users. The file size is kept under 1GB by using indexes instead of full string names. The small data set that you'll use in this notebook is a subset of the full data set. It's generally a good idea to start building a working program with a small data set to get faster performance while interacting, exploring, and getting errors with your data. When you have a fully working program, you can apply the same code to the larger data set, possibly on a larger cluster of processors. You can also minimize memory consumption by limiting the data volume as much as possible, for example, by using indexes.  Spark machine learning libraryApache Spark’s machine learning library makes practical machine learning scalable and easy. The library consists of common machine learning algorithms and utilities, including classification, regression, clustering, collaborative filtering (this notebook!), dimensionality reduction, lower-level optimization primitives, and higher-level pipeline APIs.The library has two packages:- `spark.mllib` contains the original API that handles data in RDDs. It's in maintenance mode (deprecated), but fully supported.- `spark.ml` contains a newer API for constructing ML pipelines. It handles data in DataFrames. It's being actively enhanced.You'll use the `spark.ml` package in this notebook.  Table of contents1. [Load the data](1.-Load-the-data)    1.1 [Download the data from MovieLens](1.1-Download-the-data-from-MovieLens)    1.2 [Load movies.csv](1.2-Load-movies.csv)    1.3 [Load ratings.csv](1.3-Load-ratings.csv)2. [Spark performance basics](2.-Spark-performance-basics)3. [Explore the data with Spark APIs](3.-Explore-the-data-with-Spark-APIs)4. [Visualize the data](4.-Visualize-the-data)5. [Build the recommender system](5.-Build-the-recommender-system)    5.1 [Train the model](5.1-Train-the-model)    5.2 [Run the model](5.2-Run-the-model)    5.3 [Evaluate the model](5.3-Evaluate-the-model)    5.4 [Split the data set](5.4-Split-the-data-set)    5.5 [Handle NaN results](5.5-Handle-NaN-results)    5.6 [Improve the performance score](5.6-Improve-the-performance-score)    5.7 [Improve the model](5.7-Improve-the-model)    5.8 [Recommend movies](5.8-Recommend-movies)6. [Summary and next steps](6.-Summary-and-next-steps)  1. Load the dataYou'll create Spark DataFrames, which are similar to R or pandas DataFrames, but can be distributed on a cluster of Spark executors, which can potentially scale up to thousands of machines. DataFrames are one of the easiest and best performing ways of manipulating data with Spark, but they require structured data in formats or sources such as CSV, Parquet, JSON, or JDBC. 1.1 Download the data from MovieLensTo download the data:1. Download the small version of the `ml-latest.zip` file from MovieLens Latest DataSets.1. Unzip the file. The files you'll be using are `movies.csv` and `ratings.csv`.  1.2 Load movies.csvTo load the movies.csv file into a Spark DataFrame:1. Click the **Data** button on the notebook action bar (the 1010 button at the top right of this page). Drop the files into the box or browse to select the files. The files are loaded to your object storage and also appear in the **Data Assets** section of your project.1. Click in the following code cell.1. Under the `movies.csv` file, click **Insert to code > Insert SparkSession DataFrame**. The DataFrame is named `df_data_1` by default.1. Rename the DataFrame by changing `df_data_n` to `movies` in the last row and four rows above that and run the cell.<jupyter_code># Click here to create your ratings DataFrame
import ibmos2spark<jupyter_output>Waiting for a Spark session to start...
Spark Initialization Done! ApplicationId = app-20190629202215-0001
KERNEL_ID = 04f198ca-aeb9-4a86-ac57-a473b2a419a7<jupyter_text>The ` @hidden_cell` tag hides your credentials when you use the **Share** button to share this notebook with people who are not collaborators on your project. Be aware that this tag does NOT prevent those cells from being published when you publish your notebook on GitHub!  1.3 Load ratings.csvNow load the `ratings.csv` file. Add the file to the **Files** pane and load the data in one of the following ways: -  Run **Insert to code > Insert SparkSession DataFrame** and then rename the DataFrame by changing `df_data_n` to `ratings` in the first and last row. Then run the cell. - Run the code in the next cell.Notice that there's much less code. That's because your credentials are needed only once in a notebook.<jupyter_code># Click here to create your ratings DataFrame<jupyter_output><empty_output><jupyter_text>2. Spark performance basicsSpark runs jobs in parallel on an executors cluster. You need a SparkSession instance to submit jobs. In Watson Studio, a SparkSession is automatically created, as `spark`. Show the Spark version:<jupyter_code>print(spark.version)<jupyter_output>2.3.3<jupyter_text>Show the default parallelism factor, which is the default number of partitions if you do not specify one:<jupyter_code>print(sc.defaultParallelism)<jupyter_output>2<jupyter_text>Here's how you can see the details of the Spark configuration:<jupyter_code>print(sc.getConf().toDebugString())<jupyter_output><empty_output><jupyter_text>Spark DataFrames are distributed collections across a cluster of Spark executors. One of the key factors for ensuring that a Spark job is well distributed across the cluster is the number of partitions for a DataFrame. Run the `getNumPartitions()` method to show the number of partitions for each DataFrame:<jupyter_code>print('Number of partitions for the movies DataFrame: {}'.format(movies.rdd.getNumPartitions()))
print('Number of partitions for the ratings DataFrame: {}'.format(ratings.rdd.getNumPartitions()))<jupyter_output>Number of partitions for the movies DataFrame: 1
Number of partitions for the ratings DataFrame: 1<jupyter_text>Despite the `defaultParallelism`, the Spark DataFrameReader that we used only used a single Spark partition, which means all subsequent Spark operations would only run on a single executor.You can run the `repartition()` method to redistribute the ratings data across multiple partitions. But be aware that repartition can involve a great deal of network bandwidth while data is transfered across the cluster! Because Spark DataFrames are immutable, you have to create a new DataFrame to repartition an existing DataFrame. Create the `repartitionedRatings` DataFrame across 10 partitions:<jupyter_code>repartitionedRatings = ratings.repartition(10)
print('Number of partitions for the ratings DataFrame: {}'.format(ratings.rdd.getNumPartitions()))
print('Number of partitions for the repartitionedRatings DataFrame: {}'.format(repartitionedRatings.rdd.getNumPartitions()))<jupyter_output>Number of partitions for the ratings DataFrame: 1
Number of partitions for the repartitionedRatings DataFrame: 10<jupyter_text>One of the key feature of Apache Spark is to leverage the memory as much as possible. While an action is in progress, data is loaded to memory and transformed several times before getting a result. But at the end of the processing, the memory is released. You can use the `.cache()` method to retain and persist a data set in memory as soon as you run an action on it. Load the `movies` and `ratings` data sets in memory to improve performance:<jupyter_code>print('Number of ratings: {}'.format(repartitionedRatings.count()))
repartitionedRatings.cache()<jupyter_output><empty_output><jupyter_text>When you run the next cell, `count()` is the first **action** performed on the `repartitionedRatings` DataFrame since it was cached. Thus, the data is read from object storage, the `repartition()` **transformation** is applied, and the `count()` action is run. But because you ran the `cache()` **transformation** on this DataFrame already, Spark won't release the memory.<jupyter_code>print('Number of ratings: {}'.format(repartitionedRatings.count()))<jupyter_output>Number of ratings: 100836<jupyter_text>3. Explore the data with Spark APIsYou'll use the Spark DataFrame API and SparkSQL to look at the data. The Spark DataFrame API and SparkSQL are high level APIs to query and transform Spark DataFrames. Show the content of the DataFrame in a table:<jupyter_code>movies.show(truncate=False)<jupyter_output>+-------+-------------------------------------+-------------------------------------------+
|movieId|title                                |genres                                     |
+-------+-------------------------------------+-------------------------------------------+
|1      |Toy Story (1995)                     |Adventure|Animation|Children|Comedy|Fantasy|
|2      |Jumanji (1995)                       |Adventure|Children|Fantasy                 |
|3      |Grumpier Old Men (1995)              |Comedy|Romance                             |
|4      |Waiting to Exhale (1995)             |Comedy|Drama|Romance                       |
|5      |Father of the Bride Part II (1995)   |Comedy                                     |
|6      |Heat (1995)                          |Action|Crime|Thriller                      |
|7      |Sabrina (1995)                       |Comedy|Romance                             |
|8      |Tom and Huck (1995)                  |Adventure|Children               [...]<jupyter_text>Print the schema of the DataFrame:<jupyter_code>movies.printSchema()<jupyter_output>root
 |-- movieId: string (nullable = true)
 |-- title: string (nullable = true)
 |-- genres: string (nullable = true)<jupyter_text>By default, the schema shows every column as a string. To override this, you can either explicitly specify the schema or configure the `inferSchema` parameter in the Spark CSV DataFrameReader.  Set the `inferSchema` parameter and then print the schemas.Use More option on upper right hand side panel (three dots) and insert project token.If this is the first time you are using a token in this project, you will receive message - No project access token.Below that message click on project settings and create token by clicking new token option from access tokens sections and then insert token in the notebook.The token will get added to the first cell of your notebook. Copy that token to below cell and run it.You will then be able to use the `get_file_url` method to fetch a file from the object storage using Spark.<jupyter_code># Paste here your project token.It will look like this
from project_lib import Project
project = Project(sc, '******************', '******************')
pc = project.project_context
movies_file_name = 'movies.csv'
ratings_file_name = 'ratings.csv'

movies = spark.read.csv(project.get_file_url(movies_file_name), header=True, inferSchema=True).repartition(10).cache()
ratings = spark.read.csv(project.get_file_url(ratings_file_name), header=True, inferSchema=True).repartition(10).cache()
movies.printSchema()
ratings.printSchema()<jupyter_output>root
 |-- movieId: integer (nullable = true)
 |-- title: string (nullable = true)
 |-- genres: string (nullable = true)

root
 |-- userId: integer (nullable = true)
 |-- movieId: integer (nullable = true)
 |-- rating: double (nullable = true)
 |-- timestamp: integer (nullable = true)<jupyter_text>Run the `describe()` method to see the count, mean, standard deviation, minimum, and maximum values for the data in each column:<jupyter_code>ratings.describe().show()<jupyter_output>+-------+------------------+----------------+------------------+--------------------+
|summary|            userId|         movieId|            rating|           timestamp|
+-------+------------------+----------------+------------------+--------------------+
|  count|            100836|          100836|            100836|              100836|
|   mean|326.12756356856676|19435.2957177992| 3.501556983616962|1.2059460873684695E9|
| stddev| 182.6184914634999|35530.9871987002|1.0425292390606344| 2.162610359951315E8|
|    min|                 1|               1|               0.5|           828124615|
|    max|               610|          193609|               5.0|          1537799250|
+-------+------------------+----------------+------------------+--------------------+<jupyter_text>Not all of these statistics are actually meaningful!You can use specific methods from the DataFrame API to compute any statistic:<jupyter_code>print('Number of different users: {}'.format(ratings.select('userId').distinct().count()))
print('Number of different movies: {}'.format(ratings.select('movieId').distinct().count()))
print('Number of movies with at least one rating strictly higher than 4: {}'.format(ratings.filter('rating > 4').select('movieId').distinct().count()))<jupyter_output>Number of different users: 610
Number of different movies: 9724
Number of movies with at least one rating strictly higher than 4: 4056<jupyter_text>You can also leverage your SQL knowledge to query the data. Spark version 2.0 is ANSI SQL-92 compliant and can run the 99 [TPC-DS](http://www.tpc.org/tpcds/) queries.Find the number of movies with ratings higher than 4 again, this time with SQL:<jupyter_code>ratings.createOrReplaceTempView('ratings')
spark.sql('SELECT COUNT(DISTINCT(movieId)) AS nb FROM ratings WHERE rating > 4').show()<jupyter_output>+----+
|  nb|
+----+
|4056|
+----+<jupyter_text>You can even query the data file directly, which might be ambiguous for the CSV file format, but can be useful for self-describing data formats like Parquet that embed the schema definition with the data.<jupyter_code>ratings_url = project.get_file_url(ratings_file_name)
sql = 'SELECT * FROM csv.`' + ratings_url + '`'
spark.sql(sql).take(2)<jupyter_output><empty_output><jupyter_text>You can easily switch between Spark distributed DataFrames and pandas local DataFrames.<jupyter_code>!pip install --upgrade pandas
import pandas as pd

ratings.toPandas().head()<jupyter_output><empty_output><jupyter_text>4. Visualize the dataYou'll use the Seaborn and matplotlib matplotlib libraries to create graphs. The Seaborn library works with the matplotlib library to graph statistical data.Install the Seaborn library:<jupyter_code>!pip install --upgrade seaborn<jupyter_output><empty_output><jupyter_text>Create a graph of the movies reviewed by users:<jupyter_code>import seaborn as sns
%matplotlib inline

ratingsPandas = ratings.toPandas()
sns.lmplot(x='userId', y='movieId', data=ratingsPandas, fit_reg=False);<jupyter_output><empty_output><jupyter_text>This matrix represents all the movies rated by users, but doesn't distinguish the ratings. Improve the graph by choosing a color palette to color the ratings:<jupyter_code>sns.palplot(sns.diverging_palette(10, 133, sep=80, n=10))<jupyter_output><empty_output><jupyter_text>Create the graph on a larger scale with the color palette:<jupyter_code>lm = sns.lmplot(x='userId', y='movieId', hue='rating', data=ratingsPandas, fit_reg=False, height=10, aspect=2, palette=sns.diverging_palette(10, 133, sep=80, n=10))
axes = lm.axes
axes[0, 0].set_ylim(0, 163949) # max movieId is 163949
axes[0, 0].set_xlim(0, 671) # max userId is 671
lm;<jupyter_output><empty_output><jupyter_text>On this matrix, you'll notice gaps in the data: some movies and users seem to be missing. This could be because you're using a subset of the data (the small data set).Nevertheless, you can identify some patterns. Some users always give positive reviews of movies. Some movies are rated a lot, which could be for different reasons, such as the first release of the MovieLens website had a much smaller catalog, or the movies are more famous. Now visualize the global distribution of ratings with a violin plot:<jupyter_code>sns.violinplot([ratingsPandas.rating]);<jupyter_output><empty_output><jupyter_text>These are just two examples of what you can achieve with the rich Python visualization libraries ecosystem. Feel free to explore more!  5. Build the recommender systemThere are different methods for building a recommender system, such as, user-based, content-based, or collaborative filtering. Collaborative filtering calculates recommendations based on similarities between users and products. For example, collaborative filtering assumes that users who give the similar ratings on the same movies will also have similar opinions on movies that they haven't seen.The alternating least squares (ALS) algorithm provides collaborative filtering between users and products to find products that the customers might like, based on their previous ratings.In this case, the ALS algorithm will create a matrix of all users versus all movies. Most cells in the matrix will be empty. An empty cell means the user hasn't reviewed the movie yet. The ALS algorithm will fill in the probable (predicted) ratings, based on similarities between user ratings. The algorithm uses the least squares computation to minimize the estimation errors, and alternates between solving for movie factors and solving for user factors.The following trivial example gives you an idea of the problem to solve. However, keep in mind that the general problem is much harder because the matrix often has far more missing values.![User\Product matrix](https://github.com/hatv/dsx_SparkLessons/blob/master/als.png?raw=true "User\Product matrix") Check the size of the ratings matrix:<jupyter_code>spark.sql('''
    SELECT *, 100 * nb_ratings/matrix_size AS percentage
    FROM (
        SELECT nb_users, nb_movies, nb_ratings, nb_users * nb_movies AS matrix_size
        FROM (
            SELECT COUNT(*) AS nb_ratings, COUNT(DISTINCT(movieId)) AS nb_movies, COUNT(DISTINCT(userId)) AS nb_users
            FROM ratings
        )
    )
''').toPandas().head()<jupyter_output><empty_output><jupyter_text>Less than 2% of the matrix is filled! 5.1 Train the modelUse the SparkML ALS algorithm to train a model to provide recommendations. The mandatory parameters to the ALS algorithm are the columns that identify the users, the items, and the ratings. Run the `fit()` method to train the model:<jupyter_code>from pyspark.ml.recommendation import ALS

model = ALS(userCol='userId', itemCol='movieId', ratingCol='rating').fit(ratings)<jupyter_output><empty_output><jupyter_text>5.2 Run the modelRun the `transform()` method to score the model and output a DataFrame with an additional prediction column that shows the predicted rating:<jupyter_code>predictions = model.transform(ratings)
predictions.toPandas().head()<jupyter_output><empty_output><jupyter_text>You can see that many of the predictions are close to the actual ratings. 5.3 Evaluate the modelAfter you apply a model to a data set, you should evaluate the performance of the model by comparing the predicted values with the original values. Use the RegressionEvaluator method to compare continuous values with the root mean squared calculation. The root mean squared error (RMSE) calculation measures the average of the squares of the errors between what is estimated and the existing data. The lower the mean squared error value, the more accurate the model.<jupyter_code>from pyspark.ml.evaluation import RegressionEvaluator

evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print('The root mean squared error for our model is: {}'.format(evaluator.evaluate(predictions)))<jupyter_output>The root mean squared error for our model is: 0.5932807901774939<jupyter_text>You want the performance score to improve with your design iterations so that the model is improved. But notice that you just ran the training and the scoring on the same data set. That's something that you won't normally do because you usually want to predict values that you don't already know! Therefore, this result is nonsense. To accurately evaluate the model, it's common practice in machine learning to split the data set between a training data set to train the model, and a test data set to compare the predicted results with the original results. This process is called cross-validation. Not doing cross-validation often leads to overfitting, which occurs when the model is too specific to the training data set and does not perform well on a more general data set. Here's the general iterative process of machine learning:  5.4 Split the data setSplit your ratings data set between an 80% training data set and a 20% test data set. Then rerun the steps to train the model on the training set, run it on the test set, and evaluate the performance.<jupyter_code>(trainingRatings, testRatings) = ratings.randomSplit([80.0, 20.0])
als = ALS(userCol='userId', itemCol='movieId', ratingCol='rating')
model = als.fit(trainingRatings)
predictions = model.transform(testRatings)
predictions.toPandas().head()
evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print('The root mean squared error for our model is: {}'.format(evaluator.evaluate(predictions)))<jupyter_output>The root mean squared error for our model is: nan<jupyter_text>You might get the value `nan` (not a number) from the previous cell. 5.5 Handle NaN resultsA NaN result is due to SPARK-14489 and because the model can't predict values for users for which there's no data. A temporary workaround is to exclude rows with predicted NaN values or to replace them with a constant, for instance, the general mean rating. However, to map to a real business problem, the data scientist, in collaboration with the business owner, must define what happens if such an event occurs. For example, you can provide no recommendation for a user until that user rates a few items. Alternatively, before user rates five items, you can use a user-based recommender system that's based on the user's profile (that's another recommender system to develop). Replace predicted NaN values with the average rating and evaluate the model:<jupyter_code>avgRatings = ratings.select('rating').groupBy().avg().first()[0]
print ('The average rating in the dataset is: {}'.format(avgRatings))

evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print ('The root mean squared error for our model is: {}'.format(evaluator.evaluate(predictions.na.fill(avgRatings))))<jupyter_output>The average rating in the dataset is: 3.501556983616962
The root mean squared error for our model is: 0.9000586650712746<jupyter_text>Now exclude predicted NaN values and evaluate the model:<jupyter_code>evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print ('The root mean squared error for our model is: {}'.format(evaluator.evaluate(predictions.na.drop())))<jupyter_output>The root mean squared error for our model is: 0.8863126821673949<jupyter_text>Obviously, you get lower performance than with the previous model, but you're protected against overfitting: you will actually get this level of performance on new incoming data!  5.6 Improve the performance scoreIf you run the `randomSplit()`, `fit()`, `transform()`, and `evaluate()` functions several times, you won't always get the same performance score. This is because the `randomSplit()` and `ALS()` functions have some randomness. To get a more precise performance score, run the model several times and then compute the average performance score. This process is really close to what is called ${k}$-fold cross validation. Create a `repeatALS()` function that trains, runs, and evaluates the model multiple times:<jupyter_code>def repeatALS(data, k=3, userCol='userId', itemCol='movieId', ratingCol='rating', metricName='rmse'):
    evaluations = []
    for i in range(0, k):  
        (trainingSet, testingSet) = data.randomSplit([k - 1.0, 1.0])
        als = ALS(userCol=userCol, itemCol=itemCol, ratingCol=ratingCol)
        model = als.fit(trainingSet)
        predictions = model.transform(testingSet)
        evaluator = RegressionEvaluator(metricName=metricName, labelCol='rating', predictionCol='prediction')
        evaluation = evaluator.evaluate(predictions.na.drop())
        print('Loop {}: {} = {}'.format(i + 1, metricName, evaluation))
        evaluations.append(evaluation)
    return sum(evaluations) / float(len(evaluations))<jupyter_output><empty_output><jupyter_text>Run `repeatALS` four times and average the performance scores:<jupyter_code>print('RMSE = {}'.format(repeatALS(ratings, k=4)))<jupyter_output>Loop 1: rmse = 0.8906605382403731
Loop 2: rmse = 0.8837958660975254
Loop 3: rmse = 0.8852492047370137
Loop 4: rmse = 0.8892840940492689
RMSE = 0.8872474257810452<jupyter_text>The computed performance score is more stable this way. Create a `kfoldALS()` function that also trains, runs, and evaluates the model, but splits up the data between training and testing data sets in a different way. The original data set is split into k data sets. Each of the k iterations of the function uses a different data set for testing and the other data sets for training.<jupyter_code>def kfoldALS(data, k=3, userCol='userId', itemCol='movieId', ratingCol='rating', metricName='rmse'):
    evaluations = []
    weights = [1.0] * k
    splits = data.randomSplit(weights)
    for i in range(0, k):  
        testingSet = splits[i]
        trainingSet = spark.createDataFrame(sc.emptyRDD(), data.schema)
        for j in range(0, k):
            if i == j:
                continue
            else:
                trainingSet = trainingSet.union(splits[j])
        als = ALS(userCol=userCol, itemCol=itemCol, ratingCol=ratingCol)
        model = als.fit(trainingSet)
        predictions = model.transform(testingSet)
        evaluator = RegressionEvaluator(metricName=metricName, labelCol='rating', predictionCol='prediction')
        evaluation = evaluator.evaluate(predictions.na.drop())
        print('Loop {}: {} = {}'.format(i + 1, metricName, evaluation))
        evaluations.append(evaluation)
    return sum(evaluations) / float(len(evaluations))<jupyter_output><empty_output><jupyter_text>Compute the average performance score for 4 folds:<jupyter_code>print('RMSE = {}'.format(kfoldALS(ratings, k=4)))<jupyter_output>Loop 1: rmse = 0.8807296982315512
Loop 2: rmse = 0.8948281283128832
Loop 3: rmse = 0.8863483036667585
Loop 4: rmse = 0.8856905205249661
RMSE = 0.8868991626840398<jupyter_text>Now compute the average performance score for 10 folds:<jupyter_code>print('RMSE = {}'.format(kfoldALS(ratings, k=10)))<jupyter_output>Loop 1: rmse = 0.8686338039094718
Loop 2: rmse = 0.8797971865659019
Loop 3: rmse = 0.8633948056259702
Loop 4: rmse = 0.8587820848910547
Loop 5: rmse = 0.8704924052449575
Loop 6: rmse = 0.8710890255878764
Loop 7: rmse = 0.8633550866503463
Loop 8: rmse = 0.855704172612134
Loop 9: rmse = 0.8768588912878453
Loop 10: rmse = 0.8814342315779007
RMSE = 0.8689541693953459<jupyter_text>Recall that you want to minimize the RMSE. And that 10 folds means a training set of 90% of the data, while 4 folds means a training training set of 75% of the data. If you choose a value too small for ${k}$, you will run into the selection bias issue. On the other hand, too big value for ${k}$ means overfitting - high variance and low bias. Therefore, choosing the right ${k}$ is important. Usually, the recommended value of ${k}$ is between 5 and 10, however, this can change based on the data set.  5.7 Improve the modelSo now, how can you improve this model? Machine learning algorithms have hyperparameters that control how the algorithm works.The ALS algorithm has this signature:```python    class pyspark.ml.recommendation.ALS(        rank=10,        maxIter=10,        regParam=0.1,        numUserBlocks=10,        numItemBlocks=10,        implicitPrefs=false,        alpha=1.0,        userCol="user",        itemCol="item",        seed=None,        ratingCol="rating",        nonnegative=false,        checkpointInterval=10,        intermediateStorageLevel="MEMORY_AND_DISK",        finalStorageLevel="MEMORY_AND_DISK"    )```The ALS hyperparameters are:- `rank` = the number of latent factors in the model- `maxIter` = the maximum number of iterations - `regParam` = the regularization parameterTo test several values for those **hyperparameters** and choose the best configuration, it's common practice to define a grid of parameter combinations and to run a **grid search** over the combinations to evaluate the resulting models and comparing their performance. This process is known as **model selection**. The Spark CrossValidator function performs a grid search as well as k-fold cross validation.  Run the CrossValidator function with multiple values for `rank` and `regParam`:<jupyter_code>from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

(trainingRatings, validationRatings) = ratings.randomSplit([80.0, 20.0])
evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')

paramGrid = ParamGridBuilder().addGrid(als.rank, [1, 5, 10]).addGrid(als.maxIter, [20]).addGrid(als.regParam, [0.05, 0.1, 0.5]).build()

crossval = CrossValidator(estimator=als, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=10)
cvModel = crossval.fit(trainingRatings)
predictions = cvModel.transform(validationRatings)

print('The root mean squared error for our model is: {}'.format(evaluator.evaluate(predictions.na.drop())))<jupyter_output><empty_output><jupyter_text>The more folds and parameters you add to the grid, the longer it takes to test any  combination. The `CrossValidator` model contains more information about the performance for each combination that you can get with the `avgMetrics()` method. For example, you can graph the results on a plot for analysis. Unfortunately, because of the SPARK-14489 issue mentioned above, the CrossValidator function can't compute the root mean squared error most of the time and provides incorrect results. You could limit this problem by making the training set much larger than the test set, but that's not a good practice. If you want to learn more about this issue, which is more a conceptual one than a technical one, you can have a look at <NAME>'s pull request 12896(https://github.com/apache/spark/pull/12896). Welcome to the Open Source world!  5.8 Recommend moviesTo recommend movies for a specific user, create a function that applies the trained model, `ALSModel`, on the list of movies that the user hasn't yet rated.Create a `recommendMovies` function:<jupyter_code>from pyspark.sql.functions import lit

def recommendMovies(model, user, nbRecommendations):
    # Create a Spark DataFrame with the specified user and all the movies listed in the ratings DataFrame
    dataSet = ratings.select('movieId').distinct().withColumn('userId', lit(user))

    # Create a Spark DataFrame with the movies that have already been rated by this user
    moviesAlreadyRated = ratings.filter(ratings.userId == user).select('movieId', 'userId')

    # Apply the recommender system to the data set without the already rated movies to predict ratings
    predictions = model.transform(dataSet.subtract(moviesAlreadyRated)).dropna().orderBy('prediction', ascending=False).limit(nbRecommendations).select('movieId', 'prediction')

    # Join with the movies DataFrame to get the movies titles and genres
    recommendations = predictions.join(movies, predictions.movieId == movies.movieId).select(predictions.movieId, movies.title, movies.genres, predictions.prediction)

#     recommendations.show(truncate=False)
    return recommendations<jupyter_output><empty_output><jupyter_text>Now run this function to recommend 10 movies for three different users:<jupyter_code>print('Recommendations for user 133:')
recommendMovies(model, 133, 10).toPandas()
print('Recommendations for user 471:')
recommendMovies(model, 471, 10).toPandas()
print('Recommendations for user 496:')
recommendMovies(model, 496, 10).toPandas()<jupyter_output>Recommendations for user 496:
{'fraction_non_alphanumeric': 0.09407740004276245, 'fraction_numerical': 0.031858028650844555, 'mean_word_length': 4.472171151596189, 'pattern_counts': {'":': 0, '<': 130, '<?xml version=': 0, '>': 134, 'https://': 2, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 9, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '37', 'dir': 'jupyter-structured-clean-dedup', 'id': '85689', 'license': 'MIT', 'n_tokens_mistral': 9746, 'n_tokens_neox': 8784, 'n_words': 3695, 'path': 'advanced-machine-learning and-signal-processing/Examples/Movie recommender system with Spark machine learning.ipynb', 'repo_name': 'kryvokhyzha/examples-and-courses', 'size': '32747'}