<jupyter_start><jupyter_text>Recommender Model> ***Overview***In this notebook, I use the Alternating Least Squares pipeline from Apache Spark. Then, the model with the best predictions is used to give the top ten recommended products for each user. ***Challenges and Limitations***> Ideally, this model would be used to generate recommendations from a dataset containing millions of reviews. However, without access to a distributed computing framework, I am limited to a smaller dataset using my personal laptop. > Additionally, I use the ParamGrid package to generate multiple combinations of parameters to test the model. Unfortunately, the memory available cannot handle more than a few combinations at once. To accomodate this, I will use the ParamGrid a number of times to tune the model.<jupyter_code>#Install Java and Apache Spark
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q https://www-us.apache.org/dist/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.7.tgz
!tar xf spark-2.4.5-bin-hadoop2.7.tgz
# Install spark-related depdencies for Python
!pip install -q findspark
!pip install pyspark
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.4.5-bin-hadoop2.7"
from google.colab import drive
drive.mount('/content/gdrive')
import json
import gzip
import pandas as pd

from pyspark import SparkContext
from pyspark.sql import SparkSession, SQLContext
from pyspark.sql.functions import isnan, when, count, col

import numpy as np
import functools
%matplotlib inline

from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import RegressionEvaluator
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.evaluation import RegressionEvaluator
DATA_NAME = '/content/gdrive/My Drive/lux_beauty_index.json'
APP_NAME = "Amazon Clothing Recommender"
SPARK_URL = "local[*]"
MAX_MEMORY = "6g"
sc = SparkSession.builder.appName(APP_NAME).config('spark.driver.memory', MAX_MEMORY).config('spark.driver.memory', MAX_MEMORY).master(SPARK_URL).getOrCreate()
sc.sparkContext.setCheckpointDir('/tmp')
sqlContext = SQLContext(sc)
df = sqlContext.read.json(DATA_NAME)
print(f"Dataset shape is {df.count():d} rows by {len(df.columns):d} columns.")
df.printSchema()
df.show()<jupyter_output>+----------+-------------+------+--------------+--------------+--------------------+
| productID|product_index|rating|    reviewerID|reviewer_index|               title|
+----------+-------------+------+--------------+--------------+--------------------+
|B000142FVW|          236|   4.0|A3DZFW2XL5S6XK|          1338|OPI Nail Lacquer,...|
|B0001435D4|          643|   1.0|A2LAFL37V088EU|            24|OPI Nail Lacquer ...|
|B00014GT8W|          353|   3.0|A2V92F5R7MLCVI|           326|Dermablend Cover ...|
|B00014GT8W|          353|   4.0| AS44QEHT3KSPK|           330|Dermablend Cover ...|
|B00021DXVA|          264|   5.0|A19UTUEBWKIZFT|          1947|VINCENT LONGO Wat...|
|B00021DXVA|          264|   4.0|A13KW5I9IHQ039|          3382|VINCENT LONGO Wat...|
|B00025WYZC|          786|   4.0|A2FNA5903D9E6Y|           653|Crabtree & Evelyn...|
|B0002RI2PG|           86|   5.0| AQCYEZYS51OBC|          3782|Dermablend Loose ...|
|B0002RI2PG|           86|   4.0|A2ZJPDH0YP90SD|          2773|De[...]<jupyter_text>Checking the Sparsity The ALS model builds a matrix of reviewers and products, filling in the ratings given. Many of the cells in the matrix are not filled, hence, a sparse matrix. If the matrix is too sparse, it is difficult to get accurate predictions. Having a matrix with no less than 0.05% filled is a baseline goal for the collaborative filltering method.<jupyter_code>numerator = df.count()
total_reviewers = df.select('reviewer_index').distinct().count()
total_products = df.select('product_index').distinct().count()
denominator = total_reviewers * total_products
#sparsity
sparsity = ((numerator*1.0/denominator)*100)
print('Percent of the matrix filled:')
print(sparsity)<jupyter_output>Percent of the matrix filled:
0.4730196224173525<jupyter_text>With 0.47% of the matrix filled in, we're off to a good start!  Alternating Least Squares  To begin selecting the best model, I start by building a baseline model using the default parameters. Then, to tune the model, I use a ParamGrid to try combinations of parameters.<jupyter_code>#Split the data
(training_data, test_data) = df.randomSplit([0.8, 0.2], seed=123)
#build a baseline model using default parameters
als_model = ALS(userCol='reviewer_index', itemCol='product_index', ratingCol='rating', 
                rank=10, maxIter=10, regParam= 0.1, 
                coldStartStrategy='drop', nonnegative=True, implicitPrefs=False
)
#Fit the model
base_model = als_model.fit(training_data)
#Get predictions
predictions = base_model.transform(test_data)
predictions.show()
# Complete the evaluator code
evaluator = RegressionEvaluator(metricName="rmse", labelCol="rating", predictionCol="prediction")

# Extract the 3 parameters
print(evaluator.getMetricName())
print(evaluator.getLabelCol())
print(evaluator.getPredictionCol())
# Evaluate the "predictions" dataframe
RMSE = evaluator.evaluate(predictions)

# Print the RMSE
print (RMSE)<jupyter_output>1.0561215355390317<jupyter_text>Tuning the ModelRound 1<jupyter_code>#build the model
als_model = ALS(userCol='reviewer_index', itemCol='product_index', ratingCol='rating',  
                coldStartStrategy='drop', nonnegative=True, implicitPrefs=False
)
#The parameter grid gives different combinations of parameters to test the model
param_grid = ParamGridBuilder() \
              .addGrid(als_model.rank, [10, 15]) \
              .addGrid(als_model.maxIter, [10, 15, 20]) \
              .addGrid(als_model.regParam, [.1, 0.5]) \
              .build()
#Evaluate the predictions by using rsme
evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print('Number of models to be tested: ', len(param_grid))
#Cross validation
cv = CrossValidator(estimator=als_model, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5)
print(cv)
#Fit the model
model = cv.fit(training_data)
#Extract the best combinaton of parameters
best_model = model.bestModel
test_predictions = best_model.transform(test_data)
# Calculate the RMSE of test_predictions
RMSE_1 = evaluator.evaluate(test_predictions)
print('RMSE_1 = ' + str(RMSE_1))
print('***Best Model***')
print('Rank: ', (best_model.rank))
print('MaxIter: ', (best_model._java_obj.parent().getMaxIter()))
print('RegParam: ', (best_model._java_obj.parent().getRegParam()))<jupyter_output>RMSE_1 = 1.0442751762512295
***Best Model***
Rank:  15
MaxIter:  20
RegParam:  0.1<jupyter_text>Tuning the modelRound 2 Since the local drive cannot handle more than 12 combinations, I repeated the process with increased rank and maxIter parameters to try to lower the RMSE.<jupyter_code>param_grid = ParamGridBuilder() \
              .addGrid(als_model.rank, [20, 25]) \
              .addGrid(als_model.maxIter, [10, 15, 20]) \
              .addGrid(als_model.regParam, [.1, .5]) \
              .build()
evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print('Number of models to be tested: ', len(param_grid))
cv = CrossValidator(estimator=als_model, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5)
model = cv.fit(training_data)
best_model = model.bestModel
test_predictions = best_model.transform(test_data)
RMSE_2 = evaluator.evaluate(test_predictions)
print('RMSE_2 = ' + str(RMSE_2))
print('***Best Model***')
print('Rank: ', (best_model.rank))
print('MaxIter: ', (best_model._java_obj.parent().getMaxIter()))
print('RegParam: ', (best_model._java_obj.parent().getRegParam()))<jupyter_output>RMSE_2 = 1.0372083563619778
***Best Model***
Rank:  25
MaxIter:  20
RegParam:  0.1<jupyter_text>Tuning the model: Round 3 At this point, the RMSE has gotten down to 1.033. Each round the highest parameter has been the best. Going with this pattern, I bumped up the rank and maxIter to see if I can get the RMSE down again. I've also eliminated multiple values for regParam since it hasn't made an impact in the previous combinations.<jupyter_code>param_grid = ParamGridBuilder() \
              .addGrid(als_model.rank, [30, 35, 40]) \
              .addGrid(als_model.maxIter, [20, 25, 30]) \
              .addGrid(als_model.regParam, [.1]) \
              .build()
evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print('Number of models to be tested: ', len(param_grid))
cv = CrossValidator(estimator=als_model, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5)
model = cv.fit(training_data)
best_model = model.bestModel
test_predictions = best_model.transform(test_data)

RMSE_3 = evaluator.evaluate(test_predictions)
print('RMSE_3 = ' + str(RMSE_3))
print('***Best Model***')
print('Rank: ', (best_model.rank))
print('MaxIter: ', (best_model._java_obj.parent().getMaxIter()))<jupyter_output>RMSE_3 = 1.0267772891445814
***Best Model***
Rank:  40
MaxIter:  20<jupyter_text>One more roundJust to see if we can get the RSME a any lower.<jupyter_code>param_grid = ParamGridBuilder() \
              .addGrid(als_model.rank, [50, 55 ]) \
              .addGrid(als_model.maxIter, [20, 50, 100]) \
              .addGrid(als_model.regParam, [.1]) \
              .build()
evaluator = RegressionEvaluator(metricName='rmse', labelCol='rating', predictionCol='prediction')
print('Number of models to be tested: ', len(param_grid))
cv = CrossValidator(estimator=als_model, estimatorParamMaps=param_grid, evaluator=evaluator, numFolds=5)
model = cv.fit(training_data)
best_model = model.bestModel
test_predictions = best_model.transform(test_data)

RMSE_4 = evaluator.evaluate(test_predictions)
print('RMSE_4 = ' + str(RMSE_4))
print('***Best Model***')
print('Rank: ', (best_model.rank))
print('MaxIter: ', (best_model._java_obj.parent().getMaxIter()))<jupyter_output>RMSE_4 = 1.0261489390009348
***Best Model***
Rank:  55
MaxIter:  20<jupyter_text>Model Analysis> After many iterations of the ALS model parameters, the most successful combination is a rank of 55 and maxIter of 20.  The regParam stayed consitent with the default of 0.1.  This combination yields an RMSE for product rating predictions of approximately 1.026. This means that on average, the model predicted rating score is off by about 1 star.  When looking at a 5 star rating scale, predicting anywhere from 4 to 6 stars and being off by 1 is a pretty good guess. This also means that the model could predict a user giving a 2 star rating when they actually gives a 1. In this case, the prediction is not great. However, the recommender system takes the highest predictions, so predicting 1 star too high might not be that bad.> If I had access to and the ability to process a larger amount of data, I believe this model would get an RMSE closer to zero.  This would be the benefit of using a distributed computing framework and easier access to data. > Using the best model I found, I then got the top recommendations for each user.   Get Recommendations for All Users > The recommendForAllUsers method generates a list of the n number of products with the highest predicted rating score.<jupyter_code>#Get the top 10 recommendations for all users
user_recs = best_model.recommendForAllUsers(10)
user_recs.show()<jupyter_output>+--------------+--------------------+
|reviewer_index|     recommendations|
+--------------+--------------------+
|          1580|[[1491, 4.8221908...|
|           471|[[594, 5.239127],...|
|          1591|[[594, 5.60205], ...|
|          1342|[[388, 5.2888317]...|
|          2122|[[594, 5.0728745]...|
|          2142|[[594, 5.113259],...|
|           463|[[87, 4.9177847],...|
|           833|[[666, 5.0613637]...|
|          3794|[[898, 4.829689],...|
|          1645|[[641, 5.1238627]...|
|          3175|[[594, 5.122922],...|
|           496|[[1068, 4.817307]...|
|          2366|[[236, 4.9172726]...|
|          2866|[[277, 1.6357831]...|
|           148|[[496, 4.63317], ...|
|          1088|[[875, 5.2042265]...|
|          1238|[[424, 2.3528478]...|
|          1829|[[1261, 4.7498364...|
|          1959|[[424, 5.883332],...|
|          2659|[[424, 1.1764729]...|
+--------------+--------------------+
only showing top 20 rows<jupyter_text>Improve Readability The recommendations are not in a user friendly format when first extracted. In order to see the recommendations more clearly, I used the explode and lateral view commands in sql.<jupyter_code>user_recs.createTempView('temp')
clean_recs = sqlContext.sql("SELECT reviewer_index,\
                                    products_and_ratings.product_index AS product_index,\
                                     products_and_ratings.rating AS prediction\
                              FROM temp\
                              LATERAL VIEW explode(recommendations) exploded_table\
                              AS products_and_ratings")
clean_recs.show()<jupyter_output>+--------------+-------------+----------+
|reviewer_index|product_index|prediction|
+--------------+-------------+----------+
|          1580|         1491| 4.8221908|
|          1580|          665| 4.7136793|
|          1580|         1102|   4.71349|
|          1580|         1506|  4.538516|
|          1580|         1380| 4.5076046|
|          1580|          594|  4.501891|
|          1580|          219|  4.482654|
|          1580|          560| 4.4688725|
|          1580|         1370|  4.467898|
|          1580|          661|  4.451627|
|           471|          594|  5.239127|
|           471|          577|  5.188219|
|           471|          530| 5.1852994|
|           471|          253|  5.164042|
|           471|          388| 5.1612434|
|           471|          910|  5.137744|
|           471|         1058|  5.125388|
|           471|          875|   5.12529|
|           471|          711| 5.0952144|
|           471|          784|  5.092416|
+--------------+-------------+----[...]<jupyter_text>The resulting dataframe now lists a single recommendation per row. The final steps are to add on the title of the products and convert the indices back to the original IDs.<jupyter_code>#Original dataframe containing all columns
df.show()
#Make a list of unique products with ID and index
product_info = df.select('productId', 'product_index', 'title')
unique_products = product_info.dropDuplicates()
unique_products.show()
#Add titles and productID to recommendations
recs_with_titles = clean_recs.join(unique_products, on='product_index', how='left')
recs_with_titles = recs_with_titles.drop('product_index')
recs_with_titles.show()
ratings_df = df.select('productID','reviewer_index','reviewerID','rating')
ratings_df.show()<jupyter_output>+----------+--------------+--------------+------+
| productID|reviewer_index|    reviewerID|rating|
+----------+--------------+--------------+------+
|B000142FVW|          1338|A3DZFW2XL5S6XK|   4.0|
|B0001435D4|            24|A2LAFL37V088EU|   1.0|
|B00014GT8W|           326|A2V92F5R7MLCVI|   3.0|
|B00014GT8W|           330| AS44QEHT3KSPK|   4.0|
|B00021DXVA|          1947|A19UTUEBWKIZFT|   5.0|
|B00021DXVA|          3382|A13KW5I9IHQ039|   4.0|
|B00025WYZC|           653|A2FNA5903D9E6Y|   4.0|
|B0002RI2PG|          3782| AQCYEZYS51OBC|   5.0|
|B0002RI2PG|          2773|A2ZJPDH0YP90SD|   4.0|
|B00063M3EC|          3157|A1SG1DJ72ISCRU|   2.0|
|B0006PJRRG|           786|A3HBZYCHTIGP5S|   2.0|
|B0006PLMFQ|          2854|A23H193P451P74|   3.0|
|B0006PLMFQ|           785| A7KDJ0MVYYRA0|   5.0|
|B0006PLMFQ|          1582|A17JGEYNQRYMHE|   5.0|
|B0006PLMFQ|          2442|A2P652CXM82IH6|   4.0|
|B0006PLMFQ|          2798|A34ZOFAJWL9UAM|   5.0|
|B0007M11XQ|           175| AV37OT7S682UW|   4.0|
[...]<jupyter_text>At this point, the recommendations do not account for the fact that a reviewer may have already purchased and rated the product. To address this, I joined the original review dataframe to the recommendations. If the rating is null, then the reviewer has not previously rated the product.<jupyter_code>recommendation_list = recs_with_titles.join(ratings_df, on=['productID', 'reviewer_index'], how='left_outer')
recommendation_list = recommendation_list.drop('reviewerID')
recommendation_list.show()
updated_rec_list = recommendation_list.where(col('rating').isNull())
updated_rec_list.show()
#Make a list of unique Reviewers with ID and index
reviewerID_info = df.select('reviewerId', 'reviewer_index')
unique_reviewers = reviewerID_info.dropDuplicates()
unique_reviewers.show()
#Add reviewerID to recommendations
final_rec_list = updated_rec_list.join(unique_reviewers, on='reviewer_index', how='left')
final_rec_list = final_rec_list.drop('reviewer_index')
final_rec_list.select('reviewerId', 'productId', 'title', 'prediction').show()
#To view the full product name
final_rec_list.select('reviewerId', 'productId', 'title', 'prediction').show(truncate=False)<jupyter_output>+--------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------+----------+
|reviewerId    |productId |title                                                                                                                                                       |prediction|
+--------------+----------+------------------------------------------------------------------------------------------------------------------------------------------------------------+----------+
|A2V98TS4QDDCG9|B000VOHQW0|<NAME> de To<NAME>, 4.2 fl. Oz. mens cologne                                                                                       |4.538516  |
|A2V98TS4QDDCG9|B004SUY46E|AHAVA Dead Sea Dermud Intensive Moisturizers                                                                                                                |4.5076046 |
|A2V98TS4QDDCG9|B001[...]<jupyter_text>Final ThoughtsNow we have an organized dataframe that contains the top recommended products previously unrated for each user. I personally love finding new beauty products to try. Now, to go find something awesome on Amazon!<jupyter_code><jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1530472806777808, 'fraction_numerical': 0.07149494397376333, 'mean_word_length': 2.852600547483681, 'pattern_counts': {'":': 0, '<': 40, '<?xml version=': 0, '>': 48, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 19, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '53297', 'license': 'CC-BY-3.0', 'n_tokens_mistral': 6497, 'n_tokens_neox': 5687, 'n_words': 1500, 'path': 'Final_Capstone_Model_and_Recommendations.ipynb', 'repo_name': 'ShannonElise86/ShannonElise86.gihub.io', 'size': '18320'}