<jupyter_start><jupyter_text>Spark Lens A collaborative filtering movie recommender system using the Movie Lens dataset and Spark ML.  Import libraries<jupyter_code>import pyspark
from pyspark.sql.types import *
from pyspark.ml.feature import Imputer
from pyspark.ml.recommendation import ALS
from pyspark.ml.evaluation import RegressionEvaluator

import numpy as np<jupyter_output><empty_output><jupyter_text>Initialize Spark<jupyter_code># Create Spark Session and Spark Context
spark = pyspark.sql.SparkSession \
    .builder \
    .appName("spark_lens") \
    .master("local[2]") \
    .getOrCreate()
sc = spark.sparkContext<jupyter_output><empty_output><jupyter_text>Load and clean the movie lens ratings data<jupyter_code># Load data into an RDD.
ratings_rdd = sc.textFile('data/u.data')
ratings_rdd.take(10)
# Split rows into fields.
ratings_rdd = ratings_rdd.map(lambda row: row.split('\t'))
ratings_rdd.take(10)
# Cast fields to types and drop timestamp.
def cast(row):
  user, movie, rating, timestamp = row
  return (int(user), int(movie), float(rating))
ratings_rdd = ratings_rdd.map(cast)
ratings_rdd.take(10)
# Create a schema for our Spark DataFrame.
schema = StructType([
    StructField('user', IntegerType(), True),
    StructField('movie', IntegerType(), True),
    StructField('rating', DoubleType(), True)
])
# Create our Spark DataFrame.
ratings_df = spark.createDataFrame(ratings_rdd, schema)
# Show first 10 rows.
ratings_df.show(10)
# Describe DataFrame.
ratings_df.describe().show()
# Register temp table.
ratings_df.registerTempTable('ratings')
# Query table for number of nulls.
spark.sql('''
    SELECT COUNT(rating) AS nulls
    FROM ratings 
    WHERE rating=null
''').show()<jupyter_output>+-----+
|nulls|
+-----+
|    0|
+-----+<jupyter_text>Train/test split<jupyter_code># Create a random 80/20 train/test split.
train_df, test_df = ratings_df.randomSplit([0.8, 0.2])
train_df.count()
test_df.count()<jupyter_output><empty_output><jupyter_text>Model<jupyter_code># Use Spark's implementation of an Alternating Least Squares (ALS) matrix factorization model.
als_model  = ALS( \
    userCol='user', \
    itemCol='movie', \
    ratingCol='rating', \
    nonnegative=True, \
    regParam=0.1, \
    rank=10 \
)
# Train recommender.
recommender = als_model.fit(train_df)<jupyter_output><empty_output><jupyter_text>Predict<jupyter_code># Make predictions for entire test set.
test_df = recommender.transform(test_df)
test_df.show(10)<jupyter_output>+----+-----+------+----------+
|user|movie|rating|prediction|
+----+-----+------+----------+
| 633|  148|   1.0| 3.4631267|
| 224|  148|   3.0| 3.3561032|
| 328|  148|   3.0| 3.1503906|
| 120|  148|   3.0| 2.7938714|
|  92|  148|   2.0| 2.7115803|
| 374|  148|   4.0| 3.2940223|
| 834|  148|   4.0| 3.1915097|
|  59|  148|   3.0|  2.879704|
| 244|  148|   2.0|  2.708571|
| 706|  148|   4.0| 3.4192502|
+----+-----+------+----------+
only showing top 10 rows<jupyter_text>Evaluate<jupyter_code>test_pd = test_df.toPandas()
test_pd = test_df.toPandas().fillna(test_pd['rating'].mean())
test_pd['squared_error'] = (test_pd['rating'] - test_pd['prediction'])**2
# Calculate RMSE
np.sqrt(sum(test_pd['squared_error']) / len(test_pd))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16368128474366894, 'fraction_numerical': 0.05713403335392218, 'mean_word_length': 4.3448844884488445, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, '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': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '535127', 'license': 'MIT', 'n_tokens_mistral': 1347, 'n_tokens_neox': 1203, 'n_words': 256, 'path': 'spark_lens.ipynb', 'repo_name': 'nicholsonjohnc/spark-lens', 'size': '3238'}