<jupyter_start><jupyter_text>Find the unique set of movies that have ratings in ratings_df. Remove all movies without ratings from movies_df.There are 3883 movies total, and only 3706 movies with ratings.<jupyter_code>print len(ratings_df.item_id.unique())
print len(movies_df.item_id.unique())
#26744 < 27278 so some movies do not have any ratings, these movies can be removed for genre rating averaging
# keep only movies that have at least one user rating
distinct_rated_movies = ratings_df.item_id.unique()
movies_df_rated = movies_df.iloc[[item in distinct_rated_movies for item in movies_df.item_id]]
print(movies_df_rated.shape)
del distinct_rated_movies, movies_df
tot_m, used_m, free_m = map(int, os.popen('free -t -m').readlines()[-1].split()[1:])
print 'Total memory:', tot_m
print 'Used memory:', used_m
print 'Free memory:', free_m<jupyter_output>Total memory: 517864
Used memory: 41656
Free memory: 475066<jupyter_text>Transform the ratings dataframe (user_id, item_id, label) to a user_id x item_id dataframe with the labels (ratings) as the values. Fill all missing values, items that have not been rated by a user, with 0s. Then create the binarized version of this matrix, and again fill any missing values with 0s. This dataframe will have values of 1 where a rating exists, and 0 otherwise.<jupyter_code># Transform ratings dataframe to user_id x item_id dataframe with label (or rating) as the values
ratings_spread = ratings_df.pivot(index='user_id',columns='item_id',values='label').fillna(0)
# Create binary valued dataframe from the transformed ratings dataframe. 
#   This should have values of 1 where the labels are >0 and 0 otherwise.
ratings_spread_binary = (ratings_spread/ratings_spread).fillna(0)

print('User Movie Ratings')
ratings_spread.head()<jupyter_output>User Movie Ratings<jupyter_text>Movie Genre Rating Summary<jupyter_code># Create movies dataframe subset that only contains the Movie Lens genre or IMDb genre.
genre_cols = [mg for mg in list(movies_df_rated.columns) if (mg[0:8] == 'ml_genre' or mg[0:10] == 'imdb_genre')]
movie_genres = movies_df_rated[genre_cols]
# Cross the new ratings dataframe with the genres dataframe 
#   - this is now a datframe with sum of ratings by genre for each user
user_genre_total = ratings_spread.dot(movie_genres)
# Cross the binarized ratings dataframe with the genres dataframe 
#   - this is now a dataframe that contains counts of rated movies by genre for each user
user_genre_count = ratings_spread_binary.dot(movie_genres)
ratings_spread_binary.head()
movie_genres.head()
# Perform element wise division on user genre total and user genre count to get user genre average
user_genre_avg_ratings = user_genre_total / user_genre_count
user_genre_avg_ratings = user_genre_avg_ratings.add_suffix('_avg_rating').reset_index()

# users_metadata = pd.merge(users_df, user_genre_avg_ratings, on = 'user_id')

print('User Metadata Genre Averages')
# users_metadata.head()
user_genre_avg_ratings.head()
# del movie_genres, user_genre_total, user_genre_count, user_genre_avg_ratings
del movie_genres, user_genre_total, user_genre_count<jupyter_output><empty_output><jupyter_text>Actor and Director Ratings Summary<jupyter_code># Create movies dataframe subset that only contains the actors or directors
actor_cols = [mg for mg in list(movies_df_rated.columns) if mg[0:6] == 'actor_']
director_cols = [mg for mg in list(movies_df_rated.columns) if mg[0:9] == 'director_']
people_cols = actor_cols + director_cols
movie_people = movies_df_rated[people_cols]
# # Cross the binarized ratings dataframe with the genres dataframe 
# #   - this is now a dataframe that contains counts of rated movies by genre for each user
user_movie_people_count = ratings_spread_binary.dot(movie_people)
# For testing purposes, create list of Alfred Hitchcock movies (can be done for any director)
hitchcock_movies = list(movies_df_rated.item_id.iloc[np.where(movies_df_rated.director_Alfred_Hitchcock == 1)])
print(hitchcock_movies)
# Determine how many ratings were made for Hitchcock movies
ratings_df_temp = ratings_df.iloc[np.where([item in hitchcock_movies for item in ratings_df.item_id])]
print 'number of ratings:', len(ratings_df_temp)
# Make sure that the ratings were spread correctly and that the number of ratings remains constant
print 'number of ratings:', np.sum(np.sum(ratings_spread_binary[hitchcock_movies]))
# Confirm that the dot product maintained a correct number of ratings
print 'number of ratings:', np.sum(user_movie_people_count.director_Alfred_Hitchcock)
actor_views = user_movie_people_count[actor_cols].apply(np.sum, axis = 0)
# This total will be much greater than the total number of ratings because an individual movie has multiple actors
print 'total actor views: ', np.sum(actor_views)
director_views = user_movie_people_count[director_cols].apply(np.sum, axis = 0)
# This total may be greater than the total number of ratings if one movie had multiple directors
print 'total director views: ', np.sum(director_views)
del movie_people, user_movie_people_count, hitchcock_movies, ratings_df_temp
# Filter actors to top 200, or ~7800, based on number of views
top_200_actors = actor_views.sort_values(ascending=False).head(200)
top_200_actors_names = top_200_actors.index
top_200_actors
# Filter directors to top 50, of ~2200, based on number of views
top_50_directors = director_views.sort_values(ascending=False).head(50)
top_50_directors_names = top_50_directors.index
top_50_directors
# Filter out actors and directors not in the top 200 and 50
# Create movies dataframe subset that only contains the top actors and directors
movie_people_top = movies_df_rated[list(top_200_actors_names) + list(top_50_directors_names)]
# Cross the new ratings dataframe with the  movie_people dataframes
#   - this is now a datframe with sum of ratings by actor and director for each user
user_movie_people_total = ratings_spread.dot(movie_people_top)
# Cross the binarized ratings dataframe with the movie_people dataframes
#   - this is now a dataframe with counts of rated movies by actor and director for each user
user_movie_people_count = ratings_spread_binary.dot(movie_people_top)
user_movie_people_avg_ratings = user_movie_people_total / user_movie_people_count
user_movie_people_avg_ratings = user_movie_people_avg_ratings.add_suffix('_avg_rating').reset_index()
# users_metadata = pd.merge(users_metadata, user_movie_people_avg_ratings, on = 'user_id')
users_metadata = pd.merge(user_genre_avg_ratings, user_movie_people_avg_ratings, on = 'user_id')
users_metadata.head()
del movie_people_top, user_movie_people_total, user_movie_people_count, \
    user_movie_people_avg_ratings, ratings_spread, ratings_spread_binary, \
    top_200_actors, top_50_directors, ratings_df, actor_views, \
    director_views<jupyter_output><empty_output><jupyter_text>Convert to Spark Dataframe and Save<jupyter_code>users_metadata['user_id'] = users_metadata.user_id.apply(lambda x: int(x[8:]))
sorted_columns = list(users_metadata.columns.sort_values())
users_metadata = users_metadata[sorted_columns]
users_metadata_spark = sqlContext.createDataFrame(users_metadata)
users_metadata_spark.write.format('parquet').mode('overwrite').save(data_path + 'users_metadata_20m')
movie_cols_to_keep = ['item_id', 'title', 'imdb_id', 'imdb_rating', 
                       'imdb_votes', 'metascore', 'runtime', 'year'] +\
        genre_cols + list(top_50_directors_names) + list(top_200_actors_names)
movies_df_rated['item_id'] = movies_df_rated.item_id.apply(lambda x: int(x[8:]))
movie_metadata_ohe_subset = sqlContext.createDataFrame(movies_df_rated[movie_cols_to_keep])
movie_metadata_ohe_subset.write.format('parquet').mode('overwrite').save(data_path + 'movie_20m_metadata_OHE_subset')
# genre_and_people_cols = ['item_id'] + genre_cols + list(top_50_directors_names) + list(top_200_actors_names)
# movie_genre_and_people_metadata_ohe_subset = movies_metadata[genre_and_people_cols]
# movie_genre_and_people_metadata_ohe_subset.write.format('parquet').mode('overwrite'). \
#         save(data_path + 'movie_20m_genre_and_people_metadata_ohe_subset')
sc.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.07918607501838686, 'fraction_numerical': 0.01581269919097818, 'mean_word_length': 6.991185112634672, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 20, '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': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '534259', 'license': 'Apache-2.0', 'n_tokens_mistral': 2749, 'n_tokens_neox': 2590, 'n_words': 733, 'path': 'Data_Preparation/Movie_Metadata_Preparation_and_User_Feature_Creation_20m.ipynb', 'repo_name': 'Dell-AI/Recommendation-System-in-Spark', 'size': '8158'}