<jupyter_start><jupyter_text>Setting up PySpark environment<jupyter_code>conf = SparkConf().set("spark.ui.port", "4050")
# create the context
sc = pyspark.SparkContext(conf=conf)
spark = SparkSession.builder.getOrCreate()
from google.colab import drive
drive.mount('/content/drive')
import os
cur_path = "/content/drive/MyDrive/bigdata_final"
os.chdir(cur_path)
!pwd
df = pd.read_csv("netflix_titles.csv")
df_spark = spark.read.csv("/content/drive/MyDrive/bigdata_final/netflix_titles.csv", header=True)<jupyter_output><empty_output><jupyter_text>Data preprocess<jupyter_code>from pyspark.sql.functions import lower, col<jupyter_output><empty_output><jupyter_text>Extracting useful features:<jupyter_code>df_low = df_spark.select('show_id', lower(col('director')).alias('director'),
                lower(col('description')).alias('description'),
                lower(col('title')).alias('title'),
                lower(col('listed_in')).alias('listed_in'),
                lower(col('type')).alias('type'))
df_spark.show()<jupyter_output>+-------+-------+------+--------------------+--------------------+--------------------+-----------------+------------+------+---------+--------------------+--------------------+
|show_id|   type| title|            director|                cast|             country|       date_added|release_year|rating| duration|           listed_in|         description|
+-------+-------+------+--------------------+--------------------+--------------------+-----------------+------------+------+---------+--------------------+--------------------+
|     s1|TV Show|    3%|                null|<NAME>, Bian...|              Brazil|  August 14, 2020|        2020| TV-MA|4 Seasons|International TV ...|In a future where...|
|     s2|  Movie|  7:19|   <NAME>|<NAME>, Hé...|              Mexico|December 23, 2016|        2016| TV-MA|   93 min|Dramas, Internati...|After a devastati...|
|     s3|  Movie| 23:59|        <NAME>|<NAME>...|           Singapore|December 20, 2018|     [...]<jupyter_text>Apply tokenizer function to each column<jupyter_code>from pyspark.ml.feature import Tokenizer
tokenizer = Tokenizer(inputCol="director", outputCol="director_words")
t = tokenizer.transform(df_low)
df = df.dropna(how='any')
df.drop(columns=['director', 'cast', 'country', 'date_added', 'release_year', 'rating', 'duration'], inplace=True)
import nltk
nltk.download('punkt')
nltk.download('stopwords')
from nltk.tokenize import word_tokenize

df['title_words'] = df['title'].str.lower()
df['listed_in'] = df['listed_in'].str.lower()
df['description'] = df['description'].str.lower()
df['type'] = df['type'].str.lower()

df['title_words'] = df['title_words'].apply(word_tokenize)
df['listed_in'] = df['listed_in'].apply(word_tokenize)
df['description'] = df['description'].apply(word_tokenize)
df['type'] = df['type'].apply(word_tokenize)
df
from nltk.corpus import stopwords
from string import punctuation

list_stopwords = set(stopwords.words('english') + list(punctuation))
df['title_words'] = df['title_words'].apply(lambda x: [word for word in x if word not in list_stopwords])
df['listed_in'] = df['listed_in'].apply(lambda x: [word for word in x if word not in list_stopwords])
df['description'] = df['description'].apply(lambda x: [word for word in x if word not in list_stopwords])
df['type'] = df['type'].apply(lambda x: [word for word in x if word not in list_stopwords])
import string

df['description'] = df['description'].apply(lambda x : [word.translate(str.maketrans('', '', string.punctuation)) for word in x])
df['description'] = df['description'].apply(lambda x : [word for word in x if len(word) > 0])
df['title_words'] = df['title_words'].apply(lambda x : list(set(x)))
df['listed_in'] = df['listed_in'].apply(lambda x : list(set(x)))
df['description'] = df['description'].apply(lambda x : list(set(x)))
df['type'] = df['type'].apply(lambda x : list(set(x)))
df<jupyter_output><empty_output><jupyter_text>Download the pretrained word2vec model<jupyter_code>!wget -c "https://s3.amazonaws.com/dl4j-distribution/GoogleNews-vectors-negative300.bin.gz"
!gunzip GoogleNews-vectors-negative300.bin.gz
import gensim
wv = gensim.models.KeyedVectors.load_word2vec_format("GoogleNews-vectors-negative300.bin", binary=True)
matrix_vocab = []
for list_ in df.to_numpy():
    list_[1] = [word for word in list_[1] if word in wv.vocab]
    list_[3] = [word for word in list_[3] if word in wv.vocab]
    list_[4] = [word for word in list_[4] if word in wv.vocab]
    list_[5] = [word for word in list_[5] if word in wv.vocab]
    matrix_vocab.append(list_)
df_vocab = pd.DataFrame(matrix_vocab, columns=df.columns)
from tqdm import tqdm

def recommendation(title):
    matrix_netflix_title_vocab = []
    for list_ in df[df['title'] == title].to_numpy():
        list_[1] = [word for word in list_[1] if word in wv.vocab]
        list_[3] = [word for word in list_[3] if word in wv.vocab]
        list_[4] = [word for word in list_[4] if word in wv.vocab]
        list_[5] = [word for word in list_[5] if word in wv.vocab]
        matrix_netflix_title_vocab.append(list_)

    matrix_similarity = []
    
    for list1 in matrix_vocab:
        for list2 in matrix_netflix_title_vocab:
            #score_title = wv.n_similarity(list1[5], list2[5])
            score_type = wv.n_similarity(list1[1], list2[1])
            score_catg = wv.n_similarity(list1[3], list2[3])
            score_desc = wv.n_similarity(list1[4], list2[4])
            #print(score_type)
            try:
                score_title = 0.5*wv.n_similarity(list1[5], list2[5])
            except:
                score_title = 0
            if ((list1[2] != list2[2]) & (score_catg > 0.85)):
                matrix_similarity.append([list1[2], list2[2], score_title, score_type, score_catg, score_desc])
        #pbar.update()
    #pbar.close()
    df_netflix_similarity = pd.DataFrame(matrix_similarity, columns = ['recommendation', 'title','score_type','score_title', 'score_category', 'score_description'])
    df_netflix_similarity['final_score'] = df_netflix_similarity['score_title'] + df_netflix_similarity['score_category'] + df_netflix_similarity['score_description']+ df_netflix_similarity['score_type']
    return (df_netflix_similarity.sort_values(by=['final_score', 'score_category', 'score_description', 'score_title', 'score_type'], ascending=False).head(10))<jupyter_output><empty_output><jupyter_text>Making Recommendation<jupyter_code>recommendation("Supergirl")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.19910824108241082, 'fraction_numerical': 0.017220172201722016, 'mean_word_length': 3.46771978021978, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 31, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '271112', 'license': 'MIT', 'n_tokens_mistral': 2234, 'n_tokens_neox': 2133, 'n_words': 420, 'path': 'code/recommendation.ipynb', 'repo_name': 'shiyaol/netflix_recommendation', 'size': '6544'}