<jupyter_start><jupyter_text><jupyter_code>from google.colab import drive
drive.mount('/content/drive')
!pip install pyspark

from pyspark.sql import SparkSession
from pyspark.sql.functions import count, desc, col, max
from pyspark.ml.feature import StringIndexer
from pyspark.ml import Pipeline
from pyspark.ml.recommendation import ALS
from pyspark.ml.tuning import TrainValidationSplit, ParamGridBuilder<jupyter_output><empty_output><jupyter_text>**Creating Spark Session**<jupyter_code>session1 = SparkSession.builder.appName("Spotify").getOrCreate()<jupyter_output><empty_output><jupyter_text>**Loading Dataset**<jupyter_code>file_path = '/content/drive/MyDrive/listenings.csv'
df_listenings = session1.read.format('csv').option('header', True).option('inferSchema', True).load(file_path)
df_listenings.show()<jupyter_output>+-----------+-------------+--------------------+---------------+--------------------+
|    user_id|         date|               track|         artist|               album|
+-----------+-------------+--------------------+---------------+--------------------+
|000Silenced|1299680100000|           Price Tag|       Jessie J|         Who You Are|
|000Silenced|1299679920000|Price Tag (Acoust...|       Jessie J|           Price Tag|
|000Silenced|1299679440000|Be Mine! (Ballad ...|          Robyn|            Be Mine!|
|000Silenced|1299679200000|            Acapella|          Kelis|            Acapella|
|000Silenced|1299675660000|   I'm Not Invisible|      The Tease|   I'm Not Invisible|
|000Silenced|1297511400000|Bounce (Feat NORE...|       MSTRKRFT|         Fist of God|
|000Silenced|1294498440000|Don't Stop The Mu...|        Rihanna|Addicted 2 Bassli...|
|000Silenced|1292438340000|               ObZen|      Meshuggah|               ObZen|
|000Silenced|1292437740000|   Yama's Messengers|      [...]<jupyter_text>**Data Pre-Processing**<jupyter_code># remove the date column
df_listenings = df_listenings.drop('date')
df_listenings.show()
df_listenings = df_listenings.na.drop()
df_listenings.show()
total_row = df_listenings.count()
total_column = len(df_listenings.columns)
print(total_row,total_column)<jupyter_output>13758905 4<jupyter_text>**Data Exploration**<jupyter_code>#perform sql query to dataset

df_listenings_agg = df_listenings.select('user_id','track').groupby('user_id', 'track').agg(count('*').alias('count')).orderBy('user_id')
df_listenings_agg.show()
total_rows1 = df_listenings_agg.count()
total_column1 = len(df_listenings_agg.columns)
print(total_rows1,total_column1)

#to limit the amount of data to be process to save some processing time
df_listenings_agg = df_listenings_agg.limit (20000)<jupyter_output><empty_output><jupyter_text>**Data Transformation**<jupyter_code>#convert user_id & track from string to numeric value
indexing = [StringIndexer(inputCol=col, outputCol=col+'_index').fit(df_listenings_agg)for col in list(set(df_listenings_agg.columns) - set(['count']))]

pipeline = Pipeline(stages=indexing)

data = pipeline.fit(df_listenings_agg).transform(df_listenings_agg)

data.show()
#select only 3 column from the data 

data = data.select('user_id_index' ,'track_index', 'count').orderBy('user_id_index')
data.show()<jupyter_output>+-------------+-----------+-----+
|user_id_index|track_index|count|
+-------------+-----------+-----+
|          0.0|    10943.0|    1|
|          0.0|    11628.0|    2|
|          0.0|     1349.0|    1|
|          0.0|      381.0|    1|
|          0.0|     8692.0|    1|
|          0.0|     6899.0|    1|
|          0.0|    14044.0|    1|
|          0.0|    15513.0|    1|
|          0.0|    11978.0|    2|
|          0.0|    15176.0|    1|
|          0.0|     8305.0|    1|
|          0.0|    13722.0|    1|
|          0.0|    10620.0|    1|
|          0.0|     4424.0|    1|
|          0.0|    16732.0|    1|
|          0.0|    10630.0|    1|
|          0.0|    12169.0|    1|
|          0.0|     4117.0|    1|
|          0.0|    10336.0|    1|
|          0.0|    16829.0|    1|
+-------------+-----------+-----+
only showing top 20 rows<jupyter_text>**Split Train & Test Dataset**<jupyter_code>(training,test) = data.randomSplit([0.7,0.3])<jupyter_output><empty_output><jupyter_text>**Generating the Model**<jupyter_code>User = 'user_id_index'
Track = 'track_index'
Count = 'count'

als = ALS(maxIter= 10, regParam=0.01, userCol=User, itemCol=Track, ratingCol=Count)
model = als.fit(training)

prediction = model.transform(test)<jupyter_output><empty_output><jupyter_text>**Top 20 Track Recommendation for Each User**<jupyter_code>recommend = model.recommendForAllUsers(20)
recommend.show()
#shoe the detail for 1 user
#inference = for user id 148, the system recommend track no 460 that have rating 9.16 as first recommendation
#this follow by track no 182 with rating 7.25, track 9973 rating 7.04 and so on
recommend.take(1)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1757700205338809, 'fraction_numerical': 0.07741273100616017, 'mean_word_length': 2.832415420928403, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 34, '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': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '140869', 'license': 'BSD-2-Clause', 'n_tokens_mistral': 1893, 'n_tokens_neox': 1681, 'n_words': 324, 'path': 'Recommendation_System(ALS Algorithm).ipynb', 'repo_name': 'FuadKhalit/Pyspark_Repo', 'size': '4870'}