<jupyter_start><jupyter_text>Explore MovieLens dataset using Spark<jupyter_code>import math
import os
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
# import and setup the pyspark module
from pyspark.sql import SparkSession
from pyspark.sql.types import IntegerType, StringType
from pyspark.sql import functions as func
os.environ["PYSPARK_PYTHON"] = "python3"<jupyter_output><empty_output><jupyter_text>Data ETL, Data Exploration, Data Cleaning<jupyter_code>spark = SparkSession \
            .builder \
            .appName("moive analysis") \
            .config("spark.some.config.option", "some-value") \
            .getOrCreate()
# load data
ratings = spark.read.load("/user/yg369/data/ratings.csv", format='csv', inferSchema = True, header = True)
users = spark.read.load("/user/yg369/data/users.csv", format='csv', inferSchema = True, header = True)
movies = spark.read.load("/user/yg369/data/movies.csv", format='csv', inferSchema = True, header = True)
# print data schema tree
print("ratings schema")
ratings.printSchema()
print("users schema")
users.printSchema()
print("movies schema")
movies.printSchema()
# count the number of missing values in each tags
print("Count Missing Value")

df_ls = [ratings, users, movies]
name_ls = ["ratings", "users", "movies"]

for i in range(len(df_ls)):
    print("\nFor {} data:".format(name_ls[i]))
    for column in df_ls[i].columns:
        print("Number of NULLs in column {} of {} data: ".format(column, name_ls[i]),
              df_ls[i].where(func.col(column).isNull()).count())
# since our data is complete, we don't need to clean the data<jupyter_output><empty_output><jupyter_text>Exploring Data Analysis Spark SQL and OLAP - Get Insights of Data<jupyter_code># print total number of users
print("Number of Users: ", users.select('UserID').distinct().count())

# print total number of movies
print("Number of Movies: ", movies.select('MovieID').distinct().count())
# use Spark SQL
# create data view for Spark SQL
ratings.createOrReplaceTempView("Ratings")
users.createOrReplaceTempView("Users")
movies.createOrReplaceTempView("Movies")<jupyter_output><empty_output><jupyter_text>Focus on Users.csv add users<jupyter_code># gender distribution
num_each_gender = users.select("Gender").groupby("Gender").count().withColumnRenamed("count", "Number_of_People")
num_each_gender.show()
# age distribution
age_dict = {"Under 18": "1", "18-24": "18", "25-34": "25", "35-44": "35", "45-49": "45", "50-55": "50", "56+": "56"}
age_to_dict = func.udf(lambda row: age_dict[row], StringType())

num_each_age = users.select("Age").groupby("Age").count().withColumnRenamed("count", "Number_of_People")
num_each_age = num_each_age.sort(age_to_dict("Age"))
num_each_age.show()

# plot the distribution of Number_of_People for each age
print("Plot the distribution of Number_of_People for each age:")
num_each_age_df = num_each_age.toPandas()
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="Age", y="Number_of_People", data=num_each_age_df)
display(fig)
# career distribution
num_each_occup = users.select("Occupation") \
                      .groupby("Occupation") \
                      .count().withColumnRenamed("count", "Number_of_People")
num_each_occup = num_each_occup.sort(func.col("Number_of_People").desc())
num_each_occup.show(21, truncate=False)

# plot the distribution of Number_of_People for each occupation
print("Plot the distribution of Number_of_People for each occupation:")
num_each_occup_df = num_each_occup.toPandas()
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="Number_of_People", y="Occupation", data=num_each_occup_df)
display(fig)
# average number of movies rated by one person， average ratings, could use describe
people_rated_distribution = spark.sql("SELECT Users.UserID, COUNT(Ratings.MovieID) AS Number_of_Rated_Movies, \
                                            AVG(Ratings.Rating) AS AVG_Movies_Rating \
                                       FROM Users \
                                       LEFT JOIN Ratings \
                                       ON Users.UserID = Ratings.UserID \
                                       GROUP BY Users.UserID \
                                       ORDER BY Users.UserID")
print("Total number of movies one person rated and the average rating he/she rated:")
people_rated_distribution.show(20)
people_rated_distribution_df = people_rated_distribution.toPandas()

# count of movies
dict_num_rated = dict(people_rated_distribution_df["Number_of_Rated_Movies"].describe())

print("Total number of users:\t{}".format(dict_num_rated["count"]))
print("For each user:")
print("\tAverage number of movies who rated:         \t{}".format(dict_num_rated["mean"]))
print("\tSTD of the distribution of number of movies:\t{}".format(dict_num_rated["std"]))
print("\tMinimum number of movies who rated:         \t{}".format(dict_num_rated["min"]))
print("\t25% number of movies who rated:             \t{}".format(dict_num_rated["25%"]))
print("\tMedian number of movies who rated:          \t{}".format(dict_num_rated["50%"]))
print("\t75% number of movies who rated:             \t{}".format(dict_num_rated["75%"]))
print("\tMaximum number of movies who rated:         \t{}".format(dict_num_rated["max"]))
print("\n")

# avg of ratings
dict_avg_rating = dict(people_rated_distribution_df["AVG_Movies_Rating"].describe())

print("Total number of users:\t{}".format(dict_avg_rating["count"]))
print("For each user:")
print("\tMean value of average rating who gave:         \t{}".format(dict_avg_rating["mean"]))
print("\tSTD of the distribution of average rating:\t{}".format(dict_avg_rating["std"]))
print("\tMinimum number of average rating:        \t{}".format(dict_avg_rating["min"]))
print("\t25% number of average rating:            \t{}".format(dict_avg_rating["25%"]))
print("\tMedian number of average rating:            \t{}".format(dict_avg_rating["50%"]))
print("\t75% number of average rating:            \t{}".format(dict_avg_rating["75%"]))
print("\tMaximum number of average rating:        \t{}".format(dict_avg_rating["max"]))
print("\n")

people_rated_distribution.createOrReplaceTempView("People_Profile")
#number of movies rated according to different genders, careers and ages
#ratings according to different genders, careers and ages

# gender
gender_rated_distribution = spark.sql("SELECT Users.Gender, \
                                           AVG(People_Profile.Number_of_Rated_Movies) AS AVG_Number_of_Rated_Movies, \
                                           AVG(People_Profile.AVG_Movies_Rating) AS AVG_Movies_Rating \
                                       FROM Users \
                                       LEFT JOIN People_Profile \
                                       ON Users.UserID = People_Profile.UserID \
                                       GROUP BY Users.Gender")
print("Total number of movies one person rated and the average rating with respect to gender:")
gender_rated_distribution.show()
# ages
age_dict = {"Under 18": "1", "18-24": "18", "25-34": "25", "35-44": "35", "45-49": "45", "50-55": "50", "56+": "56"}
age_to_dict = func.udf(lambda row: age_dict[row], StringType())

age_rated_distribution = spark.sql("SELECT Users.Age, \
                                        AVG(People_Profile.Number_of_Rated_Movies) AS AVG_Number_of_Rated_Movies, \
                                        AVG(People_Profile.AVG_Movies_Rating) AS AVG_Movies_Rating \
                                    FROM Users \
                                    LEFT JOIN People_Profile \
                                    ON Users.UserID = People_Profile.UserID \
                                    GROUP BY Users.Age")
age_rated_distribution = age_rated_distribution.sort(age_to_dict("Age"))
print("Total number of movies one person rated and the average rating with respect to age:")
age_rated_distribution.show()
# plot the distribution of AVG_Number_of_Rated_Movies for each age
print("Plot the distribution of AVG_Number_of_Rated_Movies for each age:")
age_rated_distribution_df = age_rated_distribution.toPandas()
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="Age", y="AVG_Number_of_Rated_Movies", data=age_rated_distribution_df)
display(fig)

# plot the distribution of AVG_Movies_Rating for each age
print("Plot the distribution of AVG_Movies_Rating for each age:")
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="Age", y="AVG_Movies_Rating", data=age_rated_distribution_df)
display(fig)
# occupation
occup_dict = {"'other' or not specified": 0, "academic/educator": 1, "artist": 2, \
              "clerical/admin": 3, "college/grad student": 4, "customer service": 5, \
              "doctor/health care": 6, "executive/managerial": 7, "farmer": 8, "homemaker": 9,\
              "K-12 student": 10, "lawyer": 11, "programmer": 12, "retired": 13,\
              "sales/marketing": 14, "scientist": 15, "self-employed": 16, "technician/engineer": 17,\
              "tradesman/craftsman": 18, "unemployed": 19, "writer": 20}
occup_to_dict = func.udf(lambda row: occup_dict[row], IntegerType())

occup_rated_distribution = spark.sql("SELECT Users.Occupation, \
                                          AVG(People_Profile.Number_of_Rated_Movies) AS AVG_Number_of_Rated_Movies, \
                                          AVG(People_Profile.AVG_Movies_Rating) AS AVG_Movies_Rating \
                                      FROM Users \
                                      LEFT JOIN People_Profile \
                                      ON Users.UserID = People_Profile.UserID \
                                      GROUP BY Users.Occupation")
# for test
# occup_rated_distribution.show()
occup_rated_distribution = occup_rated_distribution.sort(occup_to_dict("Occupation"))
print("Total number of movies one person rated and the average rating with respect to occupation:")
occup_rated_distribution.show(21)
# plot the distribution of AVG_Number_of_Rated_Movies for each occupation
print("Plot the distribution of AVG_Number_of_Rated_Movies for each occupation:")
occup_rated_distribution_df = occup_rated_distribution.toPandas()
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="AVG_Number_of_Rated_Movies", y="Occupation", data=occup_rated_distribution_df)
display(fig)

# plot the distribution of AVG_Movies_Rating for each occupation
print("Plot the distribution of AVG_Movies_Rating for each occupation:")
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="AVG_Movies_Rating", y="Occupation", data=occup_rated_distribution_df)
display(fig)<jupyter_output>Plot the distribution of AVG_Number_of_Rated_Movies for each occupation:<jupyter_text>Focus on Movies.csv<jupyter_code># count the number of rated movies
num_movies_rated = spark.sql("SELECT COUNT(DISTINCT(Movies.MovieID)) AS Number_of_Rated_Movies\
                              FROM Movies \
                              LEFT JOIN Ratings \
                              ON Movies.MovieID = Ratings.MovieID \
                              WHERE Ratings.Rating IS NOT NULL")
num_movies_rated.show(truncate=False)

# list the number of movies just rated by only 1 people
print("\n{} out of {} movies are rated by only one user\n".format(
        ratings.select("MovieID").groupby("MovieID").count().select("MovieID").where("count==1").count(), \
        movies.select('MovieID').distinct().count()))

# list the unrated movies in our data number shall be 0
movies_not_rated = spark.sql("SELECT DISTINCT(Movies.Title) AS List_of_Unrated_Movies\
                              FROM Movies \
                              LEFT JOIN Ratings \
                              ON Movies.MovieID = Ratings.MovieID \
                              WHERE Ratings.Rating IS NULL")
print("Number of movies that are not rated by users:\t{}\nFollowing is the list of not rated movies".format(
       movies_not_rated.count()))
movies_not_rated.show(truncate=False)<jupyter_output>+----------------------+
|Number_of_Rated_Movies|
+----------------------+
|3706                  |
+----------------------+


114 out of 3883 movies are rated by only one user

Number of movies that are not rated by users:	177
Following is the list of not rated movies
+-------------------------------------------------------------+
|List_of_Unrated_Movies                                       |
+-------------------------------------------------------------+
|Mondo (1996)                                                 |
|Lost Son, The (1999)                                         |
|Crows and Sparrows (1949)                                    |
|Storefront Hitchcock (1997)                                  |
|Halfmoon (<NAME> - Halbmond) (1995)                     |
|Children Are Watching us, The (Bambini ci guardano, I) (1942)|
|Love in Bloom (1935)                                         |
|Breaks, The (1999)                                           |
|<NAME> (1995)    [...]<jupyter_text>distribution of number of rating for each movies:movies_rate_distribution = spark.sql("SELECT Movies.Title, COUNT(Ratings.Rating) AS Number_of_Rating\                                      FROM Movies \                                      LEFT JOIN Ratings \                                      ON Movies.MovieID = Ratings.MovieID \                                      GROUP BY Movies.Title \                                      ORDER BY Number_of_Rating")movies_rate_distribution = movies_rate_distribution.toPandas()print("Movies with most frequent ratings: \n{}\n".format(    movies_rate_distribution.sort_values(by=["Number_of_Rating"], ascending=False).head(20)))print("\n")print("Movies with least frequent ratings: \n{}\n".format(    movies_rate_distribution.sort_values(by=["Number_of_Rating"], ascending=True).head(20)))print("\n")print("Statistical descriptive analysis on frequency of ratings for each movies:")dict_num_rating = dict(movies_rate_distribution["Number_of_Rating"].describe())print("Total number of movies:\t{}".format(dict_num_rating["count"]))print("For each movie:")print("\tAverage frequency of ratings:        \t{}".format(dict_num_rating["mean"]))print("\tSTD of the distribution of frequency:\t{}".format(dict_num_rating["std"]))print("\tMinimum frequency of ratings:        \t{}".format(dict_num_rating["min"]))print("\t25% frequency of ratings:            \t{}".format(dict_num_rating["25%"]))print("\tMedian frequency of ratings:            \t{}".format(dict_num_rating["50%"]))print("\t75% frequency of ratings:            \t{}".format(dict_num_rating["75%"]))print("\tMaximum frequency of ratings:        \t{}".format(dict_num_rating["max"]))print("\n")print("Number of movies that only have one rating:")print(movies_rate_distribution[movies_rate_distribution["Number_of_Rating"] == 1].count())print("\n")print("Number of movies that only have less than 10 rating:")print(movies_rate_distribution[movies_rate_distribution["Number_of_Rating"] < 10].count())<jupyter_code># split genres and create a new column Genre for further process
movies_split_genre = movies.withColumn("Genre", func.explode(func.split(movies.Genres, '\|')))
movies_split_genre.show(20)    # for test
# get distinct 18 Genres
genre_list = movies_split_genre.select("Genre").distinct().sort(["Genre"]).toPandas()
print("Number of Genres:\t{}\n".format(len(genre_list)))
print(genre_list)
# aggregate all the movie with same genres
movie_each_genre = movies_split_genre.groupBy("Genre").agg(func.concat_ws(", ", func.collect_list(movies["Title"])))
movie_each_genre = movie_each_genre.withColumnRenamed("concat_ws(, , collect_list(Title))", "Movie_List")
movie_each_genre = movie_each_genre.withColumn("Movie_Name", func.split("Movie_List", ","))

# count the number of movies per genres
movie_list_len = func.udf(lambda s: len(s), IntegerType())
movie_each_genre = movie_each_genre.select("Genre", "Movie_List", movie_list_len("Movie_Name").alias("Number_of_Movies"))
movie_each_genre = movie_each_genre.orderBy("Number_of_Movies", ascending=False)

# print genre with its movie list and number of movies
movie_each_genre.select("Genre", "Number_of_Movies").show()

print("Movie list for each genre")
print(movie_each_genre.toPandas())
# plot the distribution of movies for each genre
print("Plot the distribution of movies for each genre:")
movie_each_genre_df = movie_each_genre.toPandas()
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="Number_of_Movies", y="Genre", data=movie_each_genre_df)
display(fig)<jupyter_output>Plot the distribution of movies for each genre:<jupyter_text>Explore ratings.csv<jupyter_code># Statistical descriptive analysis on each genres:
movies_split_genre.createOrReplaceTempView("Movies_List")
genres_avg_rate = spark.sql("SELECT Movies_List.Genre, AVG(Ratings.Rating) AS AVG_Rating\
                             FROM Movies_List \
                             LEFT JOIN Ratings \
                             ON Movies_List.MovieID = Ratings.MovieID \
                             GROUP BY Movies_List.Genre \
                             ORDER BY AVG_Rating DESC")
genres_avg_rate.show()

genres_avg_rate = genres_avg_rate.toPandas()

dict_genres_avg_rate = dict(genres_avg_rate["AVG_Rating"].describe())
print("Total number of categories of movies:\t{}".format(dict_genres_avg_rate["count"]))
print("For each movie:")
print("\tAverage ratings for all categories:\t{}".format(dict_genres_avg_rate["mean"]))
print("\tSTD of ratings of all categories:  \t{}".format(dict_genres_avg_rate["std"]))
print("\tMinimum average ratings:           \t{}".format(dict_genres_avg_rate["min"]))
print("\t25%:                               \t{}".format(dict_genres_avg_rate["25%"]))
print("\tMedian:                               \t{}".format(dict_genres_avg_rate["50%"]))
print("\t75%:                               \t{}".format(dict_genres_avg_rate["75%"]))
print("\tMaximum average ratings:           \t{}".format(dict_genres_avg_rate["max"]))
# plot the distribution of AVG_Rating for each genre
print("Plot the distribution of AVG_Rating for each genre:")
fig = plt.figure(figsize=(13,8))
sns.set(style="whitegrid", rc={"lines.linewidth": 4.5})
sns.barplot(x="AVG_Rating", y="Genre", data=genres_avg_rate)
display(fig)<jupyter_output>Plot the distribution of AVG_Rating for each genre:<jupyter_text>Most polarising movies (Marmite movies) Next, we will try to answer the question, What are the most polarising movies? These are the movies that divide opinon, with people tending to rate them either really highly or really poorly. We will refer to these as Marmite movies. Again, we only want to consider movies that achieve some minimum number of ratings - we will stick with our previous threshold of 500 ratings.To approach this, we will look for the movies with the highest standard deviation in rating. This is a measure of how much the data varies from the mean, so in this case, how much a movie's ratings vary around its mean rating. A high standard deviation would suggest that the movie's ratings are highly variable. There are other approaches to this as well, for instance, what proportion of the ratings are very positive or very negative.<jupyter_code>ratings_stddev = ratings \
                 .groupBy("MovieID") \
                 .agg(func.count("UserID").alias("Num_of_Ratings"), 
                      func.avg(func.col("Rating")).alias("AVG_of_Rating"),
                      func.stddev(func.col("Rating")).alias("STD_of_Rating")
                     ) \
                 .where("Num_of_Ratings > 100") \
                 .sort(func.col("STD_of_Rating").desc())
# for test
# ratings_stddev.show(20)
marmite_movies = movies \
                 .join(ratings_stddev, ratings_stddev.MovieID == movies.MovieID) \
                 .sort(ratings_stddev.STD_of_Rating.desc()) \
                 .drop(ratings_stddev.MovieID)
marmite_movies.show(20)<jupyter_output>+-------+--------------------+--------------------+--------------+------------------+------------------+
|MovieID|               Title|              Genres|Num_of_Ratings|     AVG_of_Rating|     STD_of_Rating|
+-------+--------------------+--------------------+--------------+------------------+------------------+
|   1924|Plan 9 from Outer...|       Horror|Sci-Fi|           249|2.6345381526104417|1.4559983991255796|
|   2314|      Beloved (1998)|               Drama|           104|3.1346153846153846|1.3728129459672882|
|   3864|Godzilla 2000 (Go...|Action|Adventure|...|           143|2.6923076923076925|1.3646996740797475|
|   2459|Texas Chainsaw Ma...|              Horror|           247| 3.222672064777328| 1.332448440768303|
|    231|Dumb & Dumber (1994)|              Comedy|           660|3.1924242424242424|1.3213331436461841|
|   1483|        Crash (1996)|      Drama|Thriller|           141|2.7588652482269502|1.3196358221648996|
|   2710|Blair Witch Proje...|              Horror|    [...]
{'fraction_non_alphanumeric': 0.13818267419962335, 'fraction_numerical': 0.025612052730696798, 'mean_word_length': 3.0168305597579423, 'pattern_counts': {'":': 43, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 57, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '189087', 'license': 'MIT', 'n_tokens_mistral': 6908, 'n_tokens_neox': 6232, 'n_words': 1478, 'path': 'Final Exam/code/Explore MovieLens dataset using Spark.ipynb', 'repo_name': 'billgoo/Rutgers-CS543-Massive-Data-Storage-and-Retrieval', 'size': '21253'}