<jupyter_start><jupyter_text>Copyright (c) Microsoft Corporation. All rights reserved.Licensed under the MIT License.  Spark Collaborative Filtering (ALS) Deep Dive Spark MLlib provides a collaborative filtering algorithm that can be used for training a matrix factorization model, which predicts explicit or implicit ratings of users on items for recommendations.This notebook presents a deep dive into the Spark collaborative filtering algorithm.  1 Matrix factorization algorithm 1.1 Matrix factorization for collaborative filtering problemMatrix factorization is a common technique used in recommendation tasks. Basically, a matrix factorization algorithm tries to find latent factors that represent intrinsic user and item attributes in a lower dimension. That is,$$\hat r_{u,i} = q_{i}^{T}p_{u}$$where $\hat r_{u,i}$ is the predicted ratings for user $u$ and item $i$, and $q_{i}^{T}$ and $p_{u}$ are latent factors for item and user, respectively. The challenge to the matrix factorization problem is to find $q_{i}^{T}$ and $p_{u}$. This is achieved by methods such as matrix decomposition. A learning approach is therefore developed to converge the decomposition results close to the observed ratings as much as possible. Furthermore, to avoid overfitting issue, the learning process is regularized. For example, a basic form of such matrix factorization algorithm is represented as below.$$\min\sum(r_{u,i} - q_{i}^{T}p_{u})^2 + \lambda(||q_{i}||^2 + ||p_{u}||^2)$$where $\lambda$ is a the regularization parameter. In case explict ratings are not available, implicit ratings which are usually derived from users' historical interactions with the items (e.g., clicks, views, purchases, etc.). To account for such implicit ratings, the original matrix factorization algorithm can be formulated as $$\min\sum c_{u,i}(p_{u,i} - q_{i}^{T}p_{u})^2 + \lambda(||q_{i}||^2 + ||p_{u}||^2)$$where $c_{u,i}=1+\alpha r_{u,i}$ and $p_{u,i}=1$ if $r_{u,i}>0$ and $p_{u,i}=0$ if $r_{u,i}=0$. $r_{u,i}$ is a numerical representation of users' preferences (e.g., number of clicks, etc.).  1.2 Alternating Least Square (ALS)Owing to the term of $q_{i}^{T}p_{u}$ the loss function is non-convex. Gradient descent method can be applied but this will incur expensive computations. An Alternating Least Square (ALS) algorithm was therefore developed to overcome this issue. The basic idea of ALS is to learn one of $q$ and $p$ at a time for optimization while keeping the other as constant. This makes the objective at each iteration convex and solvable. The alternating between $q$ and $p$ stops when there is convergence to the optimal. It is worth noting that this iterative computation can be parallelised and/or distributed, which makes the algorithm desirable for use cases where the dataset is large and thus the user-item rating matrix is super sparse (as is typical in recommendation scenarios). A comprehensive discussion of ALS and its distributed computation can be found [here](http://stanford.edu/~rezab/classes/cme323/S15/notes/lec14.pdf).  2 Spark Mllib implementationThe matrix factorization algorithm is available as `ALS` module in [Spark `ml`](https://spark.apache.org/docs/latest/ml-collaborative-filtering.html) for DataFrame or [Spark `mllib`](https://spark.apache.org/docs/latest/mllib-collaborative-filtering.html) for RDD. * The uniqueness of ALS implementation is that it distributes the matrix factorization model training by using "Alternating Least Square" method. * In the training method, there are parameters that can be selected to control the model performance.* Both explicit and implicit ratings are supported by Spark ALS model.  3 Spark ALS based MovieLens recommenderIn the following code, the MovieLens-100K dataset is used to illustrate the ALS algorithm in Spark. **Note**: This notebook requires a PySpark environment to run properly. Please follow the steps in [SETUP.md](https://github.com/Microsoft/Recommenders/blob/master/SETUP.mddependencies-setup) to install the PySpark environment.<jupyter_code># set the environment path to find Recommenders
import sys
sys.path.append("../../")
import pandas as pd
from matplotlib import pyplot as plt
import numpy as np
import seaborn as sns
import sys
import pandas as pd

import pyspark
from pyspark.sql import SparkSession
from pyspark.ml.recommendation import ALS
import pyspark.sql.functions as F
from pyspark.sql.functions import col
from pyspark.ml.tuning import CrossValidator
from pyspark.sql.types import StructType, StructField
from pyspark.sql.types import FloatType, IntegerType, LongType

from reco_utils.dataset import movielens
from reco_utils.common.spark_utils import start_or_get_spark
from reco_utils.evaluation.spark_evaluation import SparkRankingEvaluation, SparkRatingEvaluation
from reco_utils.evaluation.parameter_sweep import generate_param_grid
from reco_utils.dataset.spark_splitters import spark_random_split

print("System version: {}".format(sys.version))
print("Pandas version: {}".format(pd.__version__))
print("PySpark version: {}".format(pyspark.__version__))<jupyter_output>System version: 3.6.0 | packaged by conda-forge | (default, Feb  9 2017, 14:36:55) 
[GCC 4.8.2 20140120 (Red Hat 4.8.2-15)]
Pandas version: 0.23.4
PySpark version: 2.3.1<jupyter_text>Data column names<jupyter_code>COL_USER = "UserId"
COL_ITEM = "MovieId"
COL_RATING = "Rating"
COL_PREDICTION = "prediction"
COL_TIMESTAMP = "Timestamp"

schema = StructType(
    (
        StructField(COL_USER, IntegerType()),
        StructField(COL_ITEM, IntegerType()),
        StructField(COL_RATING, FloatType()),
        StructField(COL_TIMESTAMP, LongType()),
    )
)<jupyter_output><empty_output><jupyter_text>Model hyper parameters - these parameters are selected with reference to the benchmarking results [here](http://mymedialite.net/examples/datasets.html).<jupyter_code>RANK = 10
MAX_ITER = 15
REG_PARAM = 0.05<jupyter_output><empty_output><jupyter_text>Number of recommended items<jupyter_code>K = 10<jupyter_output><empty_output><jupyter_text>Initialize a Spark session.<jupyter_code>spark = start_or_get_spark("ALS Deep Dive", memory="16g")<jupyter_output><empty_output><jupyter_text>3.1 Load and prepare data Data is read from csv into a Spark DataFrame.<jupyter_code>dfs = movielens.load_spark_df(spark=spark, size="100k", schema=schema)
dfs.show(5)<jupyter_output>+------+-------+------+---------+
|UserId|MovieId|Rating|Timestamp|
+------+-------+------+---------+
|   196|    242|   3.0|881250949|
|   186|    302|   3.0|891717742|
|    22|    377|   1.0|878887116|
|   244|     51|   2.0|880606923|
|   166|    346|   1.0|886397596|
+------+-------+------+---------+
only showing top 5 rows<jupyter_text>Data is then randomly split by 80-20 ratio for training and testing.<jupyter_code>dfs_train, dfs_test = spark_random_split(dfs, ratio=0.75, seed=42)<jupyter_output><empty_output><jupyter_text>3.2 Train a movielens model  It is worth noting that Spark ALS model allows dropping cold users to favor a robust evaluation with the testing data. In case there are cold users, Spark ALS implementation allows users to drop cold users in order to make sure evaluations on the prediction results are sound.<jupyter_code>als = ALS(
    maxIter=MAX_ITER, 
    rank=RANK,
    regParam=REG_PARAM, 
    userCol=COL_USER, 
    itemCol=COL_ITEM, 
    ratingCol=COL_RATING, 
    coldStartStrategy="drop"
)

model = als.fit(dfs_train)<jupyter_output><empty_output><jupyter_text>3.3 Prediction with the modelThe trained model can be used to predict ratings with a given test data.<jupyter_code>dfs_pred = model.transform(dfs_test).drop(COL_RATING)<jupyter_output><empty_output><jupyter_text>With the prediction results, the model performance can be evaluated.<jupyter_code>evaluations = SparkRatingEvaluation(
    dfs_test, 
    dfs_pred,
    col_user=COL_USER,
    col_item=COL_ITEM,
    col_rating=COL_RATING,
    col_prediction=COL_PREDICTION
)

print(
    "RMSE score = {}".format(evaluations.rmse()),
    "MAE score = {}".format(evaluations.mae()),
    "R2 score = {}".format(evaluations.rsquared()),
    "Explained variance score = {}".format(evaluations.exp_var()),
    sep="\n"
)<jupyter_output>RMSE score = 0.9667810853277086
MAE score = 0.7512176095372199
R2 score = 0.2562725605572528
Explained variance score = 0.2621805762303917<jupyter_text>Oftentimes ranking metrics are also of interest to data scientists. Note usually ranking metrics apply to the scenario of recommending a list of items. In our case, the recommended items should be different from those that have been rated by the users.<jupyter_code># Get the cross join of all user-item pairs and score them.
users = dfs_train.select('UserId').distinct()
items = dfs_train.select('MovieId').distinct()
user_item = users.crossJoin(items)
dfs_pred = model.transform(user_item)

# Remove seen items.
dfs_pred_exclude_train = dfs_pred.alias("pred").join(
    dfs_train.alias("train"),
    (dfs_pred['UserId'] == dfs_train['UserId']) & (dfs_pred['MovieId'] == dfs_train['MovieId']),
    how='outer'
)

dfs_pred_final = dfs_pred_exclude_train.filter(dfs_pred_exclude_train["train.Rating"].isNull()) \
    .select('pred.' + 'UserId', 'pred.' + 'MovieId', 'pred.' + "prediction")

dfs_pred_final.show()
evaluations = SparkRankingEvaluation(
    dfs_test, 
    dfs_pred_final,
    col_user=COL_USER,
    col_item=COL_ITEM,
    col_rating=COL_RATING,
    col_prediction=COL_PREDICTION,
    k=K
)

print(
    "Precision@k = {}".format(evaluations.precision_at_k()),
    "Recall@k = {}".format(evaluations.recall_at_k()),
    "NDCG@k = {}".format(evaluations.ndcg_at_k()),
    "Mean average precision = {}".format(evaluations.map_at_k()),
    sep="\n"
)<jupyter_output>Precision@k = 0.05254777070063696
Recall@k = 0.018860432464920123
NDCG@k = 0.04856124293537454
Mean average precision = 0.005581093384299446<jupyter_text>3.4 Fine tune the modelPrediction performance of a Spark ALS model is often affected by the parameters|Parameter|Description|Default value|Notes||-------------|-----------------|------------------|-----------------||`rank`|Number of latent factors|10|The larger the more intrinsic factors considered in the factorization modeling.||`regParam`|Regularization parameter|1.0|The value needs to be selected empirically to avoid overfitting.||`maxIters`|Maximum number of iterations|10|The more iterations the better the model converges to the optimal point.|It is always a good practice to start model building with default parameter values and then sweep the parameter in a range to find the optimal combination of parameters. The following parameter set is used for training ALS models for comparison study purposes.<jupyter_code>param_dict = {
    "rank": [10, 15, 20, 25],
    "regParam": [0.001, 0.1, 1.0]
}<jupyter_output><empty_output><jupyter_text>Generate a dictionary for each parameter combination which can then be fed into model training.<jupyter_code>param_grid = generate_param_grid(param_dict)<jupyter_output><empty_output><jupyter_text>Train models with parameters specified in the parameter grid. Evaluate the model with, for example, the RMSE metric, and then record the metrics for visualization.<jupyter_code>rmse_score = []

for g in param_grid:
    als = ALS(        
        userCol=COL_USER, 
        itemCol=COL_ITEM, 
        ratingCol=COL_RATING, 
        coldStartStrategy="drop",
        **g
    )
    
    model = als.fit(dfs_train)
    
    dfs_pred = model.transform(dfs_test).drop(COL_RATING)
    
    evaluations = SparkRatingEvaluation(
        dfs_test, 
        dfs_pred,
        col_user=COL_USER,
        col_item=COL_ITEM,
        col_rating=COL_RATING,
        col_prediction=COL_PREDICTION
    )

    rmse_score.append(evaluations.rmse())

rmse_score = [float('%.4f' % x) for x in rmse_score]
rmse_score_array = np.reshape(rmse_score, (len(param_dict["rank"]), len(param_dict["regParam"]))) 
rmse_df = pd.DataFrame(data=rmse_score_array, index=pd.Index(param_dict["rank"], name="rank"), 
                       columns=pd.Index(param_dict["regParam"], name="reg. parameter"))
fig, ax = plt.subplots()
sns.heatmap(rmse_df, cbar=False, annot=True, fmt=".4g")
display(fig)<jupyter_output><empty_output><jupyter_text>The calculated RMSE scores can be visualized to comparatively study how model performance is affected by different parameters. It can be seen from this visualization that RMSE first decreases and then increases as rank increases, due to overfitting. When the rank equals 20 and the regularization parameter equals 0.1, the model achieves the lowest RMSE score. It is noted from the visualization that the RMSE does not decrease together with increase of `rank`, which may be owing to the reason of overfitting. When `rank` is 10 and `regParam` is 0.1, the lowest RMSE score is achieved, which indicates that the model is optimal.  3.5 Top K recommendation  3.5.1 Top k for all users (items)<jupyter_code>dfs_rec = model.recommendForAllUsers(10)
dfs_rec.show(10)<jupyter_output>+------+--------------------+
|UserId|     recommendations|
+------+--------------------+
|   471|[[814, 3.5890005]...|
|   463|[[814, 3.0860794]...|
|   833|[[814, 3.2478788]...|
|   496|[[814, 3.1586995]...|
|   148|[[814, 4.0153384]...|
|   540|[[814, 3.9723701]...|
|   392|[[814, 4.169803],...|
|   243|[[814, 3.7258627]...|
|   623|[[814, 3.8202202]...|
|   737|[[814, 3.9412541]...|
+------+--------------------+
only showing top 10 rows<jupyter_text>3.5.2 Top k for a selected set of users (items)<jupyter_code>users = dfs_train.select(als.getUserCol()).distinct().limit(3)

dfs_rec_subset = model.recommendForUserSubset(users, 10)
dfs_rec_subset.show(10)<jupyter_output>+------+--------------------+
|UserId|     recommendations|
+------+--------------------+
|   471|[[814, 3.5890005]...|
|   463|[[814, 3.0860794]...|
|   148|[[814, 4.0153384]...|
+------+--------------------+<jupyter_text>3.5.3 Run-time considerations for top-k recommendationsIt is worth noting that usually computing the top-k recommendations for all users is the bottleneck of the whole pipeline (model training and scoring) of an ALS based recommendation system. This is because* Getting the top k from all user-item pairs requires a cross join which is usually very computationally expensive. * Inner products of user-item pairs are calculated individually instead of leveraging matrix block multiplication features which are available in certain contemporary computing acceleration libraries (e.g., BLAS).More details about possible optimizations of the top k recommendations in Spark can be found [here](https://engineeringblog.yelp.com/2018/05/scaling-collaborative-filtering-with-pyspark.html).<jupyter_code># cleanup spark instance
spark.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.12187351725072866, 'fraction_numerical': 0.038365078289161525, 'mean_word_length': 4.548702519744265, 'pattern_counts': {'":': 2, '<': 63, '<?xml version=': 0, '>': 64, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '17', 'dir': 'jupyter-structured-clean-dedup', 'id': '23018', 'license': 'MIT', 'n_tokens_mistral': 4957, 'n_tokens_neox': 4423, 'n_words': 1440, 'path': 'notebooks/02_model/als_deep_dive.ipynb', 'repo_name': 'ashokpant/recommenders', 'size': '14753'}