<jupyter_start><jupyter_text>Set up pyspark and SparkSession<jupyter_code>import pyspark
from pyspark.sql.types import IntegerType
from pyspark import SparkContext
import json

spark = pyspark.sql.SparkSession.builder \
    .master("local") \
    .appName("movies") \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Load data<jupyter_code><jupyter_output><empty_output><jupyter_text>Calculate average for each user<jupyter_code>averages = df \
    .groupBy("usedf = spark.read.csv(path="/home/jovyan/movielens/ratings.csv", header=True)
df = df.withColumn("rating", df["rating"].cast(IntegerType()))r_id") \
    .avg("rating") \
    .select("*")

averages.take(3)<jupyter_output><empty_output><jupyter_text>Subtract the average from each users each rating<jupyter_code>joined = averages \
    .join(df, df["user_id"] == averages["user_id"]) \
    .select(df["user_id"], "avg(rating)", "rating", "movie_id")\
    .collect()

joined[:3]<jupyter_output><empty_output><jupyter_text>Convert DataFrame to RDD<jupyter_code>sc = SparkContext.getOrCreate()
rdd = sc.parallelize(joined)<jupyter_output><empty_output><jupyter_text>Map the whole RDD to a python dictionary for easy look up<jupyter_code>def seq_op(acc, obj):
    user_id = obj["user_id"]
    movie_id = obj["movie_id"]
    average = obj["average"]
    rating = obj["rating"]
    diff = obj["diff"]
    
    if user_id not in acc:
        acc[user_id] = {
            "average": average,
            "ratings": {
                movie_id: {
                    "rating": rating,
                    "diff": diff
                }
            }
        }
    else:
        acc[user_id]["ratings"][movie_id] = {
            "rating": rating,
            "diff": diff
        }
    return acc


combOp = (lambda x, y: {**x, **y})


mapped = rdd.map(lambda row: {"user_id": row["user_id"], 
                              "rating": row["rating"],
                              "movie_id": row["movie_id"],
                              "average": row["avg(rating)"],
                              "diff": row["rating"] - row["avg(rating)"]
                              })\
            .aggregate({}, seq_op, combOp)


print(json.dumps(mapped['1'], indent=4))<jupyter_output><empty_output><jupyter_text>Define function for calculating cosine similarity<jupyter_code>import numpy as np


def cosine_similarity(first_list, second_list):

    a = np.array(first_list)
    b = np.array(second_list)
    
    dot_product = np.dot(a, b)
    
    a_length = np.linalg.norm(a)
    b_length = np.linalg.norm(b)
    
    similarity = dot_product / (a_length * b_length)
    similarity = "{0:.10f}".format(similarity)
    
    return similarity<jupyter_output><empty_output><jupyter_text>Define the compare method that compares the user to every other user<jupyter_code>def compare(user_id, mapped):
    comparison = set()
    user_movies = mapped[user_id]['ratings']

    for other_user_id in mapped:
        user = []
        other = []
        other_user_movies = mapped[other_user_id]['ratings']
        for movie_id in other_user_movies:
            if movie_id in user_movies:  # movie rated by both
                user.append(user_movies[movie_id]['diff'])
                other.append(other_user_movies[movie_id]['diff'])

        # convert ids to int for correct comparison
        user_id = int(user_id)
        other_user_id = int(other_user_id)

        smaller_id = user_id if user_id < other_user_id else other_user_id
        bigger_id = user_id if user_id > other_user_id else other_user_id

        similarity = cosine_similarity(user, other)
        comparison.add((smaller_id, bigger_id, similarity))

    return comparison<jupyter_output><empty_output><jupyter_text>Define a methods for combining sets<jupyter_code>def combine_sets(set1, set2):
    set1.update(set2)
    return set1<jupyter_output><empty_output><jupyter_text>Perform the comparison<jupyter_code>comparison = sc.parallelize(mapped) \
    .map(lambda user_id: compare(user_id, mapped)) \
    .aggregate(set(), combine_sets, combine_sets)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup