<jupyter_start><jupyter_text>CASE STUDY - Deploying a recommenderWe have seen the movie lens data on a toy dataset now lets try something a little bigger.  You have somechoices.* [MovieLens Downloads](https://grouplens.org/datasets/movielens/latest/)If your resources are limited (your working on a computer with limited amount of memory)> continue to use the sample_movielens_ranting.csvIf you have a computer with at least 8GB of RAM> download the ml-latest-small.zipIf you have the computational resources (access to Spark cluster or high-memory machine)> download the ml-latest.zipThe two important pages for documentation are below.* [Spark MLlib collaborative filtering docs](https://spark.apache.org/docs/latest/ml-collaborative-filtering.html) * [Spark ALS docs](https://spark.apache.org/docs/latest/api/python/pyspark.mllib.htmlpyspark.mllib.recommendation.ALS)<jupyter_code>import os
import shutil
import pandas as pd
import numpy as np
import pyspark as ps
from pyspark.ml import Pipeline
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.recommendation import ALS
from pyspark.sql import Row
from pyspark.sql.types import DoubleType
## ensure the spark context is available
spark = (ps.sql.SparkSession.builder
        .appName("sandbox")
        .getOrCreate()
        )

sc = spark.sparkContext
print(spark.version) 
## note that this solution uses ml-latest.zip
data_dir = os.path.join("..","data","ml-latest")
ratings_file = os.path.join(data_dir,"ratings.csv")
movies_file = os.path.join(data_dir,"movies.csv")                            
if not os.path.exists(ratings_file):
    print("ERROR make sure the path to the ratings file is correct")
## load the data
df = spark.read.format("csv").options(header="true",inferSchema="true").load(ratings_file)
df.show(n=4)
movies_df = pd.read_csv(movies_file)
movies_df.rename(columns={"movieId": "movie_id"},inplace=True)
movies_df.head(5)<jupyter_output><empty_output><jupyter_text>QUESTION 1Explore the movie lens data a little and summarize it<jupyter_code>## YOUR CODE HERE (summarize the data)
df = df.withColumnRenamed("movieID", "movie_id")
df = df.withColumnRenamed("userID", "user_id")
df.describe().show()
print('Unique users: {}'.format(df.select('user_id').distinct().count()))
print('Unique movies: {}'.format(df.select('movie_id').distinct().count()))
print('Movies with Rating > 2: {}'.format(df.filter('rating > 2').select('movie_id').distinct().count()))
print('Movies with Rating > 3: {}'.format(df.filter('rating > 3').select('movie_id').distinct().count()))
print('Movies with Rating > 4: {}'.format(df.filter('rating > 4').select('movie_id').distinct().count()))<jupyter_output>+-------+------------------+-----------------+------------------+--------------------+
|summary|           user_id|         movie_id|            rating|           timestamp|
+-------+------------------+-----------------+------------------+--------------------+
|  count|          27753444|         27753444|          27753444|            27753444|
|   mean|141942.01557064414|18487.99983414671|3.5304452124932677|1.1931218549319258E9|
| stddev| 81707.40009148757|35102.62524746828| 1.066352750231982|2.1604822852233925E8|
|    min|                 1|                1|               0.5|           789652004|
|    max|            283228|           193886|               5.0|          1537945149|
+-------+------------------+-----------------+------------------+--------------------+

Unique users: 283228
Unique movies: 53889
Movies with Rating > 2: 50735
Movies with Rating > 3: 43107
Movies with Rating > 4: 29374<jupyter_text>QUESTION 2Find the ten most popular movies---that is the then movies with the highest average rating>Hint: you may want to subset the movie matrix to only consider movies with a minimum number of ratings<jupyter_code>## YOUR CODE HERE

## get the top rated movies with more than 100 ratings
movie_counts = df.groupBy("movie_id").count()
top_rated = df.groupBy("movie_id").avg('rating')
top_rated = top_rated.withColumnRenamed("movie_id", "movie_id_2")
top_movies = top_rated.join(movie_counts, top_rated.movie_id_2 == movie_counts.movie_id)
top_movies = top_movies.filter('count>100').orderBy('avg(rating)',ascending=False).drop("movie_id_2")
top_movies = top_movies.toPandas()

## add the movie titles to data frame
movie_ids = top_movies['movie_id'].values
inds = [np.where(movies_df['movie_id'].values==mid)[0][0] for mid in movie_ids]
top_movies["title"] = movies_df['title'].values[inds]

top_movies.head(10)<jupyter_output><empty_output><jupyter_text>QUESTION 3Compare at least 5 different values for the ``regParam``Use the `` ALS.trainImplicit()`` and compare it to the ``.fit()`` method.  See the [Spark ALS docs](https://spark.apache.org/docs/latest/api/python/pyspark.mllib.htmlpyspark.mllib.recommendation.ALS)for example usage.<jupyter_code>## YOUR CODE HERE

(training, test) = df.randomSplit([0.8, 0.2])

def train_model(reg_param,implicit_prefs=False):
    als = ALS(maxIter=5, regParam=reg_param, userCol="user_id", 
              itemCol="movie_id", ratingCol="rating",
              coldStartStrategy="drop",implicitPrefs=implicit_prefs)
    model = als.fit(training)

    predictions = model.transform(test)
    evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating",
                                    predictionCol="prediction")

    rmse = evaluator.evaluate(predictions)
    print("regParam={}, RMSE={}".format(reg_param,np.round(rmse,2)))

for reg_param in [0.01, 0.05, 0.1, 0.15, 0.25]:
    train_model(reg_param)<jupyter_output>regParam=0.01, RMSE=0.84
regParam=0.05, RMSE=0.83
regParam=0.1, RMSE=0.83
regParam=0.15, RMSE=0.84
regParam=0.25, RMSE=0.87<jupyter_text>QUESTION 4With your best regParam try using the `implicitPrefs` flag.>Note that the results here make sense because the data are `explicit` ratings<jupyter_code>## YOUR CODE HERE

train_model(0.1, implicit_prefs=True)<jupyter_output>regParam=0.1, RMSE=3.26<jupyter_text>QUESTION 5Use model persistence to save your finalized model<jupyter_code>## YOUR CODE HERE

## re-train using the whole data set
print("...training")
als = ALS(maxIter=5, regParam=0.1, userCol="user_id", 
          itemCol="movie_id", ratingCol="rating",
          coldStartStrategy="drop")
model = als.fit(df)

## save the model for furture use
save_dir = "saved-recommender"
if os.path.isdir(save_dir):
    print("...overwritting saved model")
    shutil.rmtree(save_dir)

## save the top-ten movies
print("...saving top-movies")
top_movies[:10000].to_csv("top-movies.csv",index=False)
    
## save model
model.save(save_dir)
print("done.")<jupyter_output>...training
...overwritting saved model
...saving top-movies
done.<jupyter_text>QUESTION 6Use ``spark-submit`` to load the model and demonstrate that you can load the model and interface with it.<jupyter_code>## YOUR CODE HERE

## see recommender-submit.py<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16116000577117298, 'fraction_numerical': 0.046890780551147024, 'mean_word_length': 3.8577435178696566, 'pattern_counts': {'":': 1, '<': 25, '<?xml version=': 0, '>': 40, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '588779', 'license': 'Apache-2.0', 'n_tokens_mistral': 2511, 'n_tokens_neox': 2226, 'n_words': 524, 'path': 'AI-Workflow-Enterprise-Model-Deployment/Model_Deployment-case-study_V3/m5-case-study-soln.ipynb', 'repo_name': 'tallamjr/ibm-advanced-data-science', 'size': '6931'}