<jupyter_start><jupyter_text>Goal: Song Popularity Prediction using non-linear models Name: <NAME> Machine Learning Algo: Decision Tree and Random Forest<jupyter_code>from pyspark import SparkContext
from pyspark.sql.window import Window
from pyspark.sql import SparkSession 
from pyspark.sql.types import *
from pyspark.sql.functions import *
from pyspark.ml.feature import StringIndexer

from pyspark.sql import SparkSession
from pyspark.sql.types import *

ss = SparkSession.builder.getOrCreate()
sc = ss.sparkContext
spark<jupyter_output><empty_output><jupyter_text>Reading Data<jupyter_code>database = 'oms'
collection = 'song_popularity_data'
user_name = 'oms'
password = '<PASSWORD>'
address = 'oms-cluster.0navm.mongodb.net'
connection_string = f"mongodb+srv://{user_name}:{password}@{address}/{database}.{collection}"
df_oms = spark.read.format("mongo").option("uri",connection_string).load()
%%time
df_oms.show(4)<jupyter_output><empty_output><jupyter_text>CPU times: user 28.2 ms, sys: 12.8 ms, total: 41 msWall time: 11.5 s  Train Test SplitTake all non null values as train and do prediction on the values where "song_hotttnesss" is Null<jupyter_code>df_test = df_oms.filter(df_oms['song_hotttnesss'].isNull())
df_train = df_oms.filter(df_oms['song_hotttnesss'].isNotNull())
%%time
df_test.count()
df_train.count()<jupyter_output><empty_output><jupyter_text>CPU times: user 30 ms, sys: 11.2 ms, total: 41.2 msWall time: 11.7 s<jupyter_code>### Select needed columns
df_train = df_train.select('duration','start_of_fade_out', 'time_signature','tempo','key','mode','year','loudness','song_hotttnesss')
df_test = df_test.select('duration','start_of_fade_out', 'time_signature','tempo','key','mode','year','loudness','song_hotttnesss')
df_test.cache()
df_train.cache()<jupyter_output><empty_output><jupyter_text>train valdiation split<jupyter_code>data_split = df_train.randomSplit([0.8,0.2], seed = 10)

train_data = data_split[0]
val_data = data_split[1]

train_data.cache()
val_data.cache()
train_data.display()<jupyter_output><empty_output><jupyter_text>String-Indexingonly year column needs string indexing<jupyter_code>si = StringIndexer(inputCol='year', outputCol="year-num")
sm = si.fit(train_data)
train_data_num = sm.transform(train_data).drop('year')
valid_data_num = sm.transform(val_data).drop('year')
df_test_num = sm.transform(df_test).drop('year')

train_data_num = train_data_num.withColumnRenamed("year-num", "year").cache()
valid_data_num = valid_data_num.withColumnRenamed("year-num", "year").cache()
df_test_num = df_test_num.withColumnRenamed("year-num", "year").cache()
%%time
train_data_num.display()<jupyter_output><empty_output><jupyter_text>CPU times: user 480 µs, sys: 54 µs, total: 534 µsWall time: 620 µs<jupyter_code>train_data_num.groupBy('year').count().show()<jupyter_output><empty_output><jupyter_text>Vector Assembler<jupyter_code>from pyspark.ml.feature import VectorAssembler
cols = ['duration', 'start_of_fade_out', 'time_signature', 'tempo', 'key', 'mode', 'loudness', 'year']
va = VectorAssembler(outputCol="features", inputCols=cols) 
train_data_vec = va.transform(train_data_num).select("features", "song_hotttnesss").withColumnRenamed('song_hotttnesss', 'label')
valid_data_vec = va.transform(valid_data_num).select("features", "song_hotttnesss").withColumnRenamed('song_hotttnesss', 'label')
test_data_vec = va.transform(df_test_num).select("features", "song_hotttnesss").withColumnRenamed('song_hotttnesss', 'label')
%%time
train_data_vec.display()<jupyter_output><empty_output><jupyter_text>Decision Tree Regressor with cross-validationmaxDepth: [5,10,15,20,25,30] increasing depth allows more deep tree, i.e. more bushy trees\maxBins: [16,32,48] increasing maxBins allow more finer splitsEvaluation Metric: RMSE<jupyter_code>from pyspark.ml.regression import DecisionTreeRegressor
from pyspark.ml.tuning import CrossValidator
from pyspark.ml.tuning import ParamGridBuilder
from pyspark.ml.evaluation import RegressionEvaluator


evaluator = RegressionEvaluator().setLabelCol("label")\
                .setPredictionCol("prediction")\
                .setMetricName('rmse')

dt = DecisionTreeRegressor()
paramGrid = ParamGridBuilder().addGrid(dt.maxDepth, [5,10,15,20,25,30]).addGrid(dt.maxBins, [16,32,48]).build()

cv = CrossValidator(estimator=dt, 
                      evaluator=evaluator, 
                      numFolds=3, 
                      estimatorParamMaps=paramGrid)





%%time
cvmodel = cv.fit(train_data_vec)
%%time
dtpredicts = cvmodel.bestModel.transform(valid_data_vec)

print("Best Model Max depth : %s" % cvmodel.bestModel.getMaxDepth())
print("Best Model Max bins : %s" % cvmodel.bestModel.getMaxBins())
print("RMSE : %s" % evaluator.evaluate(dtpredicts))<jupyter_output><empty_output><jupyter_text>Random Forest Model<jupyter_code>from pyspark.ml.regression import RandomForestRegressor

%%time
rf = RandomForestRegressor(seed = 2, maxDepth=20)
rf_model = rf.fit(train_data_vec)
print(rf_model.toDebugString)<jupyter_output><empty_output><jupyter_text>RF Model time :5.14 minutes<jupyter_code>rfpredicts = rf_model.transform(valid_data_vec)
print("RMSE : %s" % evaluator.evaluate(rfpredicts))
rf_model.getNumTrees<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13339712918660288, 'fraction_numerical': 0.016076555023923446, 'mean_word_length': 7.71, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 47, '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': '11', 'dir': 'jupyter-structured-clean-dedup', 'id': '567216', 'license': 'MIT', 'n_tokens_mistral': 1937, 'n_tokens_neox': 1825, 'n_words': 306, 'path': 'ML_notebooks/Ashwani_Song_Popularity.ipynb', 'repo_name': 'shubhamthakur01/Building-SpotFlix', 'size': '5225'}