<jupyter_start><jupyter_text>**Item Based Collaborative Filtering**This notebook explains how to use the Amazon Reviews Dataset to build a item-based recommender using latent factorized collaborative filtering with Spark's Alternating Least Saqures (ALS) implementation. It is organised in two parts:The first one is about getting and parsing titles and ratings data into Spark 1.   Visualizing the preformance of ALS based on input parameters2.   Given an item of choice, recommending relevant items> This notebook gives an inference about similar products, given products of choice.  ***Setting up the pyspark envirnoment***<jupyter_code># Install pyspark and initiate
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q https://downloads.apache.org/spark/spark-2.4.5/spark-2.4.5-bin-hadoop2.6.tgz
!tar xf spark-2.4.5-bin-hadoop2.6.tgz
!pip install -q findspark
#Envirnoment set up
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-2.4.5-bin-hadoop2.6"
#Spark session creation
import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local[*]").getOrCreate()
#Import libraries
import time

# spark imports
from pyspark.sql import SparkSession
from pyspark.sql.types import *
from pyspark.sql import Row
from pyspark.sql.functions import UserDefinedFunction, explode, desc
from pyspark.sql.types import StringType, ArrayType
from pyspark.mllib.recommendation import ALS
from pyspark.ml.feature import StringIndexer

# data science imports
import math
import numpy as np
import pandas as pd

# visualization imports
import seaborn as sns
import matplotlib.pyplot as plt

%matplotlib inline
#Creating an object to run SparkContext and configuration parameters set up
sc = spark.sparkContext
spark.conf.set("spark.executor.memory", '8g')
spark.conf.set('spark.executor.cores', '3')
spark.conf.set('spark.cores.max', '3')
spark.conf.set("spark.driver.memory",'8g')
#Drive mounting 
from google.colab import drive
drive.mount('/content/drive')<jupyter_output>Go to this URL in a browser: https://accounts.google.com/o/oauth2/auth?client_id=947318989803-6bn6qk8qdgf4n4g3pfee6491hc0brc4i.apps.googleusercontent.com&redirect_uri=urn%3aietf%3awg%3aoauth%3a2.0%3aoob&response_type=code&scope=email%20https%3a%2f%2fwww.googleapis.com%2fauth%2fdocs.test%20https%3a%2f%2fwww.googleapis.com%2fauth%2fdrive%20https%3a%2f%2fwww.googleapis.com%2fauth%2fdrive.photos.readonly%20https%3a%2f%2fwww.googleapis.com%2fauth%2fpeopleapi.readonly

Enter your authorization code:
··········
Mounted at /content/drive<jupyter_text>***Analyzing the ratings dataset***<jupyter_code>#Creating a RDD
data = spark.read \
    .format("csv") \
    .option("header", "true") \
    .option("inferSchema", "true") \
    .load("/content/drive/My Drive/AmazonRatings")
data.withColumn('userID', data.userID.cast(IntegerType()))
data.withColumn('productID', data.productID.cast(IntegerType()))
data.withColumn('rating', data.rating.cast(FloatType()))
#Summary of the data
data.describe().show()
#Understanding the schema of the data
data.printSchema()
print('Distinct values of ratings:')
print(sorted(data.select('rating').distinct().rdd.map(lambda r: r[0]).collect()))
temp1 = data.groupBy("userID").count().toPandas()['count'].min()
temp2 = data.groupBy("productID").count().toPandas()['count'].min()
print('For the users that rated items and the items that were rated:')
print('Minimum number of ratings per user is {}'.format(temp1))
print('Minimum number of ratings per item is {}'.format(temp2))
count = data.select('userID').distinct().count()
print('Total number of distinct users are: {} in the data'.format(count))
count = data.select('productID').distinct().count()
print('Total number of distinct items are: {} in the data'.format(count))<jupyter_output>Total number of distinct items are: 8852 in the data<jupyter_text>**Steps for building recommendation model using ALS***   Load the data as a spark RDD dataframe*   Split the data into train, test and validation dataset (6:2:2)*   Train the ALS model with train data using different input parameters:        - Number of iterations        - Number of latent factors        - Number of regularizations ***Load the Spark RDD data***<jupyter_code>ratings = sc.textFile('/content/drive/My Drive/AmazonRatings')

# preprocess data -- only need ["userId", "productId", "rating"]
header = ratings.take(1)[0]
rating_data = ratings \
    .filter(lambda line: line!=header) \
    .map(lambda line: line.split(",")) \
    .map(lambda tokens: (int(tokens[0]), int(tokens[1]), float(tokens[2]))) \
    .cache()
rating_data.take(4)<jupyter_output><empty_output><jupyter_text>***Train-Validation-Test Split***<jupyter_code>train, validation, test = rating_data.randomSplit([6, 2, 2], seed=99)
# cache data
train.cache()
validation.cache()
test.cache()
# Each dataset size
print('Train dataset size: ', train.count())
print('Validation dataset size: ', validation.count())
print('Test dataset size: ' ,test.count())<jupyter_output>Train dataset size:  29829
Validation dataset size:  10149
Test dataset size:  10022<jupyter_text>***Training the model with different input parameters***<jupyter_code>def train_ALS(train_data, validation_data, num_iters, reg_param, ranks):
    """
    Grid Search Function to select the best model based on RMSE of hold-out data
    """
    # initial
    min_error = float('inf')
    best_rank = -1
    best_regularization = 0
    best_model = None
    for rank in ranks:
        for reg in reg_param:
            # train ALS model
            model = ALS.train(
                ratings=train_data,    # (userID, productID, rating) tuple
                iterations=num_iters,
                rank=rank,
                lambda_=reg,           # regularization param
                nonnegative = True,
                seed=99)
            # make prediction
            valid_data = validation_data.map(lambda p: (p[0], p[1]))
            predictions = model.predictAll(valid_data).map(lambda r: ((r[0], r[1]), r[2]))
            # get the rating result
            ratesAndPreds = validation_data.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
            # get the RMSE
            MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).mean()
            error = math.sqrt(MSE)
            print('{} latent factors and regularization = {}: validation RMSE is {}'.format(rank, reg, error))
            if error < min_error:
                min_error = error
                best_rank = rank
                best_regularization = reg
                best_model = model
    print('\nThe best model has {} latent factors and regularization = {}'.format(best_rank, best_regularization))
    return best_model
# hyper-param config
num_iterations = 5
ranks = [10, 12, 14, 16, 18, 20]
reg_params = [0.001, 0.01, 0.05, 0.1, 0.2]

# grid search and select best model
start_time = time.time()
final_model = train_ALS(train, validation, num_iterations, reg_params, ranks)
print ('Total Runtime: {:.2f} seconds'.format(time.time() - start_time))<jupyter_output>10 latent factors and regularization = 0.001: validation RMSE is 3.8862582022966667
10 latent factors and regularization = 0.01: validation RMSE is 2.0685580467217615
10 latent factors and regularization = 0.05: validation RMSE is 1.6366131301314546
10 latent factors and regularization = 0.1: validation RMSE is 1.542165048422527
10 latent factors and regularization = 0.2: validation RMSE is 1.4617201012741474
12 latent factors and regularization = 0.001: validation RMSE is 3.3721834947478047
12 latent factors and regularization = 0.01: validation RMSE is 2.0127573632264166
12 latent factors and regularization = 0.05: validation RMSE is 1.6598461080290603
12 latent factors and regularization = 0.1: validation RMSE is 1.5756098141965846
12 latent factors and regularization = 0.2: validation RMSE is 1.4938474923659772
14 latent factors and regularization = 0.001: validation RMSE is 3.60193085063171
14 latent factors and regularization = 0.01: validation RMSE is 1.9024970227996185
14 laten[...]<jupyter_text>***Plotting the model for different number of iterations***<jupyter_code>def plot_learning_curve(arr_iters, train_data, validation_data, reg, rank):
    """
    Plot function to show learning curve of ALS
    """
    errors = []
    for num_iters in arr_iters:
        # train ALS model
        model = ALS.train(
            ratings=train_data,    # (userID, productID, rating) tuple
            iterations=num_iters,
            rank=rank,
            lambda_=reg,           # regularization param
            nonnegative = True,
            seed=99)
        # make prediction
        valid_data = validation_data.map(lambda p: (p[0], p[1]))
        predictions = model.predictAll(valid_data).map(lambda r: ((r[0], r[1]), r[2]))
        # get the rating result
        ratesAndPreds = validation_data.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
        # get the RMSE
        MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).mean()
        error = math.sqrt(MSE)
        # add to errors
        errors.append(error)

    # plot
    plt.figure(figsize=(12, 6))
    plt.plot(arr_iters, errors, marker='s',c='g')
    plt.xlabel('number of iterations')
    plt.ylabel('RMSE')
    plt.title('ALS Learning Curve')
    plt.grid(True)
    plt.show()
# create an array of num_iters
iter_array = list(range(1, 20))
# create learning curve plot
plot_learning_curve(iter_array, train, validation, 0.2, 10)<jupyter_output><empty_output><jupyter_text>***Plotting the model for different number of latent factors***<jupyter_code>def plot_learning_curve(ranks, train_data, validation_data, reg, iters):
    """
    Plot function to show learning curve of ALS
    """
    errors = []
    for rank in ranks:
        # train ALS model
        model = ALS.train(
            ratings=train_data,    # (userID, productID, rating) tuple
            iterations=iters,
            rank=rank,
            lambda_=reg,           # regularization param
            nonnegative = True,
            seed=99)
        # make prediction
        valid_data = validation_data.map(lambda p: (p[0], p[1]))
        predictions = model.predictAll(valid_data).map(lambda r: ((r[0], r[1]), r[2]))
        # get the rating result
        ratesAndPreds = validation_data.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
        # get the RMSE
        MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).mean()
        error = math.sqrt(MSE)
        # add to errors
        errors.append(error)

    # plot
    plt.figure(figsize=(12, 6))
    plt.plot(ranks, errors,marker='s',c='r')
    plt.xlabel('number of latent factors')
    plt.ylabel('RMSE')
    plt.title('ALS Learning Curve')
    plt.grid(True)
    plt.show()
# create an array of latent factors
rank_array = [8,10,12,14,16,18,20]
# create learning curve plot
plot_learning_curve(rank_array, train, validation, 0.01, 15)<jupyter_output><empty_output><jupyter_text>***Evaluation on test data***<jupyter_code># make prediction using test data
test_data = test.map(lambda p: (p[0], p[1]))
predictions = final_model.predictAll(test_data).map(lambda r: ((r[0], r[1]), r[2]))
# get the rating result
ratesAndPreds = test.map(lambda r: ((r[0], r[1]), r[2])).join(predictions)
# get the RMSE
MSE = ratesAndPreds.map(lambda r: (r[1][0] - r[1][1])**2).mean()
error = math.sqrt(MSE)
print('The out-of-sample RMSE of rating predictions is', round(error, 4))
#Plotting the predictions
predictions.take(20)<jupyter_output><empty_output><jupyter_text>***Predictions and analysis of best model***<jupyter_code>#original vs predictions
ratesAndPreds.take(20)
final_model.userFeatures().sortByKey().take(2)
final_model.productFeatures().sortByKey().take(2)
np.dot( 
    final_model.userFeatures().sortByKey().take(1)[0][1], 
    final_model.productFeatures().sortByKey().take(1)[0][1] 
)
product = sc.parallelize([(0, 500)]) 
individual_rating_RDD = final_model.predictAll(product)
individual_rating_RDD.collect()<jupyter_output><empty_output><jupyter_text>**Making Recommendations of Products**<jupyter_code>#reload the data
ratings = sc.textFile('/content/drive/My Drive/AmazonRatings')

# preprocess data -- only need ["userId", "productId", "rating"]
header = ratings.take(1)[0]
rating_data = ratings \
    .filter(lambda line: line!=header) \
    .map(lambda line: line.split(",")) \
    .map(lambda tokens: (int(tokens[0]), int(tokens[1]), float(tokens[2]))) \
    .cache()
#Checking the tuple
rating_data.take(10)
#Load the titles data
pd.options.display.max_colwidth = 100
title_df = pd.read_csv('/content/drive/My Drive/AmazonTitles')
#Drop the duplicates in titles data
title_df.drop_duplicates(subset = ['productID'],inplace = True)
#Total products
print('Number of unique products: ', title_df.shape[0])
#Display of productID's with their respective titles
pd.options.display.max_colwidth = 100
title_df.head(20)
#Create a spark dataframe
titles = spark.createDataFrame(title_df)<jupyter_output><empty_output><jupyter_text>> **get_itemId**: This function returns the productsID's corresponding to user's choice of products>> 1.   Input parameters: title data and user's item list2.   Return parameter: list of productID's correspponding to user choice<jupyter_code>def get_itemId(df_titles, mylist):
    itemID_list = []
    for item in mylist:
        itemIds = df_titles \
            .filter(titles.title.like('%{}%'.format(item))) \
            .select('productID') \
            .rdd \
            .map(lambda r: r[0]) \
            .collect()
        itemID_list.extend(itemIds)
    return list(set(itemID_list))<jupyter_output><empty_output><jupyter_text>> **add_data_to_train**: This function returns the new training data including the new user to the existing training data>> 1.   Input parameters: training data and user's item list2.   Return parameter: new training data<jupyter_code>def add_data_to_train(train_data, item_list):
    # get new user id
    new_id = train_data.map(lambda r: r[0]).max() + 1
    # get max rating
    max_rating = train_data.map(lambda r: r[2]).max()
    # create new user rdd
    user_rows = [(new_id, item, max_rating) for item in item_list]
    new_rdd = sc.parallelize(user_rows)
    # return new train data
    return train_data.union(new_rdd)<jupyter_output><empty_output><jupyter_text>> **get_modified**: This function returns the inference data where we exclude the items of users choice and then make predictions>> 1.   Input parameters: training data, title data and user's item list2.   Return parameter: test data for prediction<jupyter_code>def get_modified(train_data, title_data, item_list):
    # get new user id
    new_id = train_data.map(lambda r: r[0]).max() + 1
    # return inference rdd
    return title_data.rdd \
        .map(lambda r: r[0]) \
        .distinct() \
        .filter(lambda x: x not in item_list) \
        .map(lambda x: (new_id, x))<jupyter_output><empty_output><jupyter_text>***Making Recommendations by training the best model***<jupyter_code>def get_recommendations(best_model_params, ratings_data, title_data, 
                        mylist,toprecs):
    # modify train data by adding new user's rows
    prod_list = get_itemId(title_data, mylist)
    train_data = add_data_to_train(ratings_data, prod_list)
    
    # train best ALS
    model = ALS.train(
        ratings=train_data,
        iterations=best_model_params.get('iterations', None),
        rank=best_model_params.get('rank', None),
        lambda_=best_model_params.get('lambda_', None),
        seed=99)
    
    # get inference rdd
    inference_rdd = get_modified(ratings_data, title_data, prod_list)
    
    # inference
    predictions = model.predictAll(inference_rdd).map(lambda r: (r[1], r[2]))
    
    # get top n productids
    topn_rows = predictions.sortBy(lambda r: r[1], ascending=False).take(toprecs)
    topn_ids = [r[0] for r in topn_rows]
    
    # return titles for top products
    return title_data.filter(title_data.productID.isin(topn_ids)) \
                    .select('title') \
                    .rdd \
                    .map(lambda r: r[0]) \
                    .collect()<jupyter_output><empty_output><jupyter_text>A) Test for a single item from the item list(title data) available<jupyter_code># my items
my_item_list = ['Zojirushi NHS-10 6-Cup (Uncooked) Rice Cooker']

# get recommends
recommends = get_recommendations(
    best_model_params={'iterations': 15, 'rank': 10, 'lambda_': 0.2}, 
    ratings_data=rating_data, 
    title_data=titles, 
    mylist=my_item_list,
    toprecs=10)

print('Recommendations for {}:'.format(my_item_list[0]))
for i, title in enumerate(recommends):
    print('{0}: {1}'.format(i+1, title))
# my items
my_item_list = ['Cuisipro Stainless Steel Measuring Spoon Set']

# get recommends
recommends = get_recommendations(
    best_model_params={'iterations': 15, 'rank': 10, 'lambda_': 0.2}, 
    ratings_data=rating_data, 
    title_data=titles, 
    mylist=my_item_list,
    toprecs=20)

print('Recommendations for {}:'.format(my_item_list[0]))
for i, title in enumerate(recommends):
    print('{0}: {1}'.format(i+1, title))<jupyter_output>Recommendations for Cuisipro Stainless Steel Measuring Spoon Set:
1: Polder 307T 307 Preset Thermometer, White
2: Le Creuset Enameled Cast Iron Round French Oven (5 1/2 Quart, Cherry Red)
3: Eureka 570B  Cordless Boss Upright Vacuum
4: Hamilton Beach 04162 HEPA Air Purifier with UV Germicidal Light
5: Coopers DIY Beer Home Brewing 6 Gallon All Inclusive Craft Beer Making Kit with Patented Brewing Fermenter, Beer Hydrometer, Brewing Ingredients, Bottles and Brewing Accessories
6: Medelco 4 Cup Universal Glass Espresso Replacement Carafe, Black
7: Anolon Advanced Nonstick Hard-Anodized Aluminum 8-Piece Cookware Set
8: Cuisipro Accutec Razor-Sharp Kitchen Rasp
9: Prepac Black 7-Drawer Lingerie Chest
10: The ARC Curved Shower Bar, Stainless Steel
11: Vacu Vin Foil Cutter with Magnetic Holder
12: EnviroCare Replacement Micro Filtration Vacuum Bags for Hoover Windtunnel Upright Type Y 18 Pack
13: Polder 893-90 Dual Kitchen 20-Hour Timer/Clock/Stopwatch
14: KitchenAid Silicone Mixer Spatula, [...]<jupyter_text>B) Test for 2 items from the item list(title data) available<jupyter_code># my items
my_item_list = ['Proctor-Silex 26500Y Durable Belgian Waffle Baker',
                'OXO Good Grips Garlic Press']

# get recommends
recommends = get_recommendations(
    best_model_params={'iterations': 15, 'rank': 10, 'lambda_': 0.2}, 
    ratings_data=rating_data, 
    title_data=titles, 
    mylist=my_item_list,
    toprecs = 20)

print('Recommendations for {}:'.format(my_item_list))
for i, title in enumerate(recommends):
    print('{0}: {1}'.format(i+1, title))<jupyter_output>Recommendations for ['Proctor-Silex 26500Y Durable Belgian Waffle Baker', 'OXO Good Grips Garlic Press']:
1: Amco Stainless-Steel Cocktail Picks, Set of 4
2: Pyrex 6022369 Storage 14-Piece Round Set, Clear with Blue Lids
3: FoodSaver 1 Quart Size Bag, Package of 48
4: simplehuman KT1025 System Stainless-Steel Dish Rack
5: KitchenAid KHM900ER 9-Speed Hand-Mixer, Empire Red
6: Classic 7" Hollow Edge Santoku Knife with Board
7: MASTERCOOL (52224-A-SP Gray Infrared Thermometer with Laser
8: Polder 77-90 Diet Utility Food Scale, 18 oz./500 g Capacity, White
9: OXO Good Grips Vegetable Brush (color may vary)
10: William Bounds Sili Gourmet V-Shaped Roasting Rack
11: Melitta Coffee Maker, 6 Cup Pour-Over Brewer with Glass Carafe, 1-Count
12: Bardwil Reflections Spill Proof Oblong / Rectangle Tablecloth, 60-Inch x 102-Inch, Pearl
13: Cuisinart TOB-195 Exact Heat Toaster Oven Broiler, Stainless
14: Lodge 10-1/4-Inch Cast-Iron Lid
15: Weber 7514 Warm-Up Basket (24.3 x 4 x 0.6)
16: Brentwood Orig[...]
{'fraction_non_alphanumeric': 0.11756409610638191, 'fraction_numerical': 0.037324333853825616, 'mean_word_length': 4.066088287828528, 'pattern_counts': {'":': 0, '<': 60, '<?xml version=': 0, '>': 70, 'https://': 2, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '16', 'dir': 'jupyter-structured-clean-dedup', 'id': '331659', 'license': 'Apache-2.0', 'n_tokens_mistral': 7033, 'n_tokens_neox': 6389, 'n_words': 1790, 'path': 'ALS Recommendation System/Step4_ALS Collaborative Filtering (Item Based).ipynb', 'repo_name': 'Thaslim/CMPE_255_Amzn', 'size': '19853'}