<jupyter_start><jupyter_text>Sparkify Project WorkspaceThis workspace contains a tiny subset (128MB) of the full dataset available (12GB). Feel free to use this workspace to build your project, or to explore a smaller subset with Spark before deploying your cluster on the cloud. Instructions for setting up your Spark cluster is included in the last lesson of the Extracurricular Spark Course content.You can follow the steps below to guide your data analysis and model building portion of this project.<jupyter_code># Import libraries
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
from pyspark.sql.types import IntegerType
from pyspark.ml.feature import VectorAssembler, Normalizer, StandardScaler
from pyspark.ml.classification import LogisticRegression, RandomForestClassifier, GBTClassifier, DecisionTreeClassifier, NaiveBayes, LinearSVC
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder

from pyspark.sql.functions import avg, col, concat, desc, explode, lit, min, max, sum, split, udf, isnan, when, count, countDistinct
import time
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

# create a Spark session
spark = SparkSession \
    .builder \
    .appName("Sparkify") \
    .getOrCreate()
#check installed spark version
print('Version: ', spark.version)<jupyter_output>Version:  2.4.3<jupyter_text>Load and Clean DatasetIn this workspace, the mini-dataset file is `mini_sparkify_event_data.json`. Load and clean the dataset, checking for invalid or missing data - for example, records without userids or sessionids.<jupyter_code>file = "mini_sparkify_event_data.json"
df = spark.read.json(file)

#reducing memory
df.persist()
#a glance of the data
df.head()
df.printSchema()
df.columns
#count null value from each column
for column in df.columns:
    print(column, "missing values:", df.filter(F.isnull(df[column])).count())<jupyter_output>artist missing values: 58392
auth missing values: 0
firstName missing values: 8346
gender missing values: 8346
itemInSession missing values: 0
lastName missing values: 8346
length missing values: 58392
level missing values: 0
location missing values: 8346
method missing values: 0
page missing values: 0
registration missing values: 8346
sessionId missing values: 0
song missing values: 58392
status missing values: 0
ts missing values: 0
userAgent missing values: 8346
userId missing values: 0<jupyter_text>Check for missing values in string form since from the lecture, it is mentioned that the missing values are in the form of string<jupyter_code>#checking missing values in the form of string ("")
for column in df.columns:
    print(column, "missing values (str):", df.filter(df[column] == '').count())<jupyter_output>artist missing values (str): 0
auth missing values (str): 0
firstName missing values (str): 0
gender missing values (str): 0
itemInSession missing values (str): 0
lastName missing values (str): 0
length missing values (str): 0
level missing values (str): 0
location missing values (str): 0
method missing values (str): 0
page missing values (str): 0
registration missing values (str): 0
sessionId missing values (str): 0
song missing values (str): 0
status missing values (str): 0
ts missing values (str): 0
userAgent missing values (str): 0
userId missing values (str): 8346<jupyter_text>From above, we see that there are some missing values for userID in string form. UserID and sessionID are the primary data points that we will use, thus it is important for it to be filled.<jupyter_code>df_clean = df.dropna(how='any', subset=['userId', 'sessionId'])
df_clean = df_clean.filter(df_clean.userId != '')
# check if there's still missing values of userID
print('userId empty string values:', df_clean.filter(df_clean.userId == '').count())
# clean data points
df_clean.count()
# descriptive statistics for user id
df_clean.describe('userId').show()
# descriptive statistics for session id
df_clean.describe('sessionID').show()
# descriptive statistics df clean
df_clean.describe('*').show()
# checking the date time
df_clean.select("ts","registration").show()
# Change ts and registration timestamp format to DateType format
df_clean = df_clean.withColumn('ts', F.from_unixtime(F.col('ts') / 1000))
df_clean = df_clean.withColumn('registration', F.from_unixtime(F.col('registration') / 1000))
df_clean.select("ts","registration").show()<jupyter_output>+-------------------+-------------------+
|                 ts|       registration|
+-------------------+-------------------+
|2018-10-01 00:01:57|2018-09-28 22:22:42|
|2018-10-01 00:03:00|2018-09-30 18:20:30|
|2018-10-01 00:06:34|2018-09-28 22:22:42|
|2018-10-01 00:06:56|2018-09-30 18:20:30|
|2018-10-01 00:11:16|2018-09-28 22:22:42|
|2018-10-01 00:11:18|2018-09-30 18:20:30|
|2018-10-01 00:14:46|2018-09-30 18:20:30|
|2018-10-01 00:14:59|2018-09-28 22:22:42|
|2018-10-01 00:15:05|2018-09-28 22:22:42|
|2018-10-01 00:18:04|2018-09-28 22:22:42|
|2018-10-01 00:19:06|2018-09-30 18:20:30|
|2018-10-01 00:19:10|2018-09-30 18:20:30|
|2018-10-01 00:20:18|2018-09-28 22:22:42|
|2018-10-01 00:22:55|2018-09-30 18:20:30|
|2018-10-01 00:22:56|2018-09-30 18:20:30|
|2018-10-01 00:24:01|2018-09-28 22:22:42|
|2018-10-01 00:26:16|2018-09-30 18:20:30|
|2018-10-01 00:27:48|2018-09-19 13:53:39|
|2018-10-01 00:28:07|2018-09-28 22:22:42|
|2018-10-01 00:29:04|2018-09-30 18:20:30|
+-------------------+-------------[...]<jupyter_text>Exploratory Data AnalysisWhen you're working with the full dataset, perform EDA by loading a small subset of the data and doing basic manipulations within Spark. In this workspace, you are already provided a small subset of data you can explore. Define ChurnOnce you've done some preliminary analysis, create a column `Churn` to use as the label for your model. I suggest using the `Cancellation Confirmation` events to define your churn, which happen for both paid and free users. As a bonus task, you can also look into the `Downgrade` events. Explore DataOnce you've defined churn, perform some exploratory data analysis to observe the behavior for users who stayed vs users who churned. You can start by exploring aggregates on these two groups of users, observing how much of a specific action they experienced per a certain time unit or number of songs played.<jupyter_code>df_clean.show(10)
#using sparksql for EDA
#Create a temporary view to perform SQL operations
user_log = spark.read.json("mini_sparkify_event_data.json")

user_log.createOrReplaceTempView("sparkify")

#sqlContext.setConf("spark.sql.tungsten.enabled", "false")<jupyter_output><empty_output><jupyter_text>Pre-eliminary analysis of users as well as cancellation confirmation and downgrade as the indicator of churning<jupyter_code># explore by page
spark.sql('''
        SELECT 
        page,
        COUNT((userID)) AS count_user
        FROM sparkify
        GROUP BY page
        ORDER BY 2 DESC
''').show()
# unique number of users 
spark.sql('''
        SELECT 
        COUNT(DISTINCT(userID))
        FROM sparkify
''').show()
# number of users who has "Cancellation Confirmation"
spark.sql('''
    SELECT
    COUNT (DISTINCT userID)
    FROM sparkify
    WHERE page = 'Cancellation Confirmation'
''').show()
# number of users who has "Downgrade"
spark.sql('''
    SELECT
    COUNT (DISTINCT userID)
    FROM sparkify
    WHERE page = 'Downgrade'
''').show()<jupyter_output>+----------------------+
|count(DISTINCT userID)|
+----------------------+
|                   154|
+----------------------+<jupyter_text>Gender Distribution<jupyter_code>gender = spark.sql('''
    SELECT
    COUNT (DISTINCT userID) AS count_user,
    gender
    FROM sparkify
    GROUP BY gender
''').show()<jupyter_output>+----------+------+
|count_user|gender|
+----------+------+
|       104|     F|
|         1|  null|
|       121|     M|
+----------+------+<jupyter_text>Level Distribution (Paid or Not)<jupyter_code>spark.sql('''
    SELECT
    userID,
    level
    FROM sparkify
    --GROUP BY level
''').show()<jupyter_output>+------+-----+
|userID|level|
+------+-----+
|    30| paid|
|     9| free|
|    30| paid|
|     9| free|
|    30| paid|
|     9| free|
|     9| free|
|    30| paid|
|    30| paid|
|    30| paid|
|     9| free|
|     9| free|
|    30| paid|
|     9| free|
|     9| free|
|    30| paid|
|     9| free|
|    74| free|
|    30| paid|
|     9| free|
+------+-----+
only showing top 20 rows<jupyter_text>From the above, we can assume that there are 136 users who changed from free to paid  Most popular songs played<jupyter_code>spark.sql('''
    SELECT
    COUNT(song),
    song
    FROM sparkify
    GROUP BY song
    ORDER BY 1 DESC
''').show(n = 10)<jupyter_output>+-----------+--------------------+
|count(song)|                song|
+-----------+--------------------+
|       1153|      You're The One|
|       1026|                Undo|
|        854|             Revelry|
|        728|       Sehr kosmisch|
|        641|Horn Concerto No....|
|        574|Dog Days Are Over...|
|        466|             Secrets|
|        459|        Use Somebody|
|        435|              Canada|
|        424|             Invalid|
+-----------+--------------------+
only showing top 10 rows<jupyter_text>Most popular artist listened by users<jupyter_code>spark.sql('''
    SELECT
    artist,
    COUNT (artist)
    FROM sparkify
    GROUP BY artist
    ORDER BY 2 DESC
''').show(n = 10)<jupyter_output>+--------------------+-------------+
|              artist|count(artist)|
+--------------------+-------------+
|       <NAME>|         1841|
|            Coldplay|         1813|
|Florence + The Ma...|         1236|
|       <NAME>|         1135|
|            BjÃÂ¶rk|         1133|
|      The Black Keys|         1125|
|                Muse|         1090|
|       <NAME>|         1044|
|        <NAME>|         1007|
|              Eminem|          953|
+--------------------+-------------+
only showing top 10 rows<jupyter_text>Item in session<jupyter_code>spark.sql('''
    SELECT
    userId,
    sessionId,
    (itemInSession)
    FROM sparkify

''').show(n = 10)<jupyter_output>+------+---------+-------------+
|userId|sessionId|itemInSession|
+------+---------+-------------+
|    30|       29|           50|
|     9|        8|           79|
|    30|       29|           51|
|     9|        8|           80|
|    30|       29|           52|
|     9|        8|           81|
|     9|        8|           82|
|    30|       29|           53|
|    30|       29|           54|
|    30|       29|           55|
+------+---------+-------------+
only showing top 10 rows<jupyter_text>Length of Song<jupyter_code>spark.sql('''
    SELECT
    userId,
    sessionId,
    length,
    song,
    page
    FROM sparkify
''').show(n = 10)<jupyter_output>+------+---------+---------+--------------------+---------------+
|userId|sessionId|   length|                song|           page|
+------+---------+---------+--------------------+---------------+
|    30|       29|277.89016|           Rockpools|       NextSong|
|     9|        8|236.09424|              Canada|       NextSong|
|    30|       29| 282.8273|   Time For Miracles|       NextSong|
|     9|        8|262.71302|Knocking On Forbi...|       NextSong|
|    30|       29|223.60771|Harder Better Fas...|       NextSong|
|     9|        8|208.29995|      Don't Leave Me|       NextSong|
|     9|        8|260.46649|         Run Run Run|       NextSong|
|    30|       29|185.44281|Passengers (Old A...|       NextSong|
|    30|       29|     null|                null|Add to Playlist|
|    30|       29|134.47791|          Fuck Kitty|       NextSong|
+------+---------+---------+--------------------+---------------+
only showing top 10 rows<jupyter_text>The definition of churn that I am going to use in this project is the one who has Cancellation Confirmation as it is more like a definite indicator<jupyter_code># create a churn column, classify it from the cancellation confirmation 
churn = F.udf(lambda page: 1 if page == 'Cancellation Confirmation' else 0, IntegerType())
df_churn = df_clean.withColumn('churn_user', churn(df_clean.page))
df_churn = df_churn.groupBy('userId').agg(F.sum('churn_user').alias('churn_group'))
df_churn.show()
# joining the churn column to the original column
df_clean = df_clean.join(df_churn, ['userId'])
df_clean.show(n = 5)
df_clean.printSchema()
# plot how many users who curn and not
classify_churn = df_clean.select("userId", "churn_group").distinct().groupBy("churn_group").count().toPandas()
classify_churn['churn_group'] = ["no", "yes"]
classify_churn
sns.barplot(x ="churn_group", y = "count", data = classify_churn)
plt.title('Number of Churned Users')
plt.xlabel('Churned or Not')
plt.ylabel('Total User')
plt.show()<jupyter_output><empty_output><jupyter_text>Exploratory Data Analysis for Churned and Not Churned  Gender and Churned<jupyter_code>gender_churn = df_clean.select("userId", "gender", "churn_group").distinct().groupBy("churn_group", "gender").count().toPandas()
gender_churn
# selecting the gender column
gender_df = df_clean.select('userId', 'gender').distinct()
# converting the gender into integer type with lambda function
gender_categories = F.udf(lambda gender: 0 if gender == 'M' else 1, IntegerType())
# adding gender categories as a new column
gender_df = gender_df.withColumn('gender', gender_categories(F.col('gender')))
# joining gender column to the model colum
df_model = df_model.join(gender_df, ['userId'])
# plot the comparison of male and female who churned
sns.barplot(x='churn_group', y='count', hue = 'gender', data = gender_churn)
plt.title('Number of Churned Users by Gender')
plt.xlabel('Churned or Not')
plt.ylabel('Total User')
plt.show()<jupyter_output><empty_output><jupyter_text>From the graph above, more male is both churned and not churned compared to female. In males, churn rate is 0.26 which is 0.19 in females  Paid or Not and Churned<jupyter_code>level_churn = df_clean.dropDuplicates(["userId"]).groupby(["churn_group", "level"]).count().toPandas()
level_churn
# plot the comparison of paid and free users who churned
sns.barplot(x='churn_group', y='count', hue = 'level', data = level_churn)
plt.title('Number of Churned Users by Paid and Free Users')
plt.xlabel('Churned or Not')
plt.ylabel('Total User')
plt.show()<jupyter_output><empty_output><jupyter_text>The ratio of paid users who churn is 0.17 while for free users it is 0.25. From the graph above, the ration of free users are higher both in churn and not churned.  Overview of the event on the page column<jupyter_code>df_page = df_clean.select(["page","churn_group"]).groupby([ "page","churn_group"]).count().sort("page").toPandas()
sns.barplot( x="count",y="page", hue="churn_group", data=df_page)<jupyter_output><empty_output><jupyter_text>Number of Songs played - Churned and Not Churned Users<jupyter_code># number of songs played is indicated when user is clicking nextsong page
# below is the total number of songs played (not with distinct users) since dropping duplicates users will reduce a lot of songs as well that the user play
song_churn = df_clean.where(df_clean['page'] == 'NextSong').groupby(["churn_group"]).count().toPandas()
song_churn
song_avg = df_clean.where(df_clean['page'] == 'NextSong').groupby("churn_group").agg((F.count("churn_group")/F.countDistinct("userId")).alias('avgsong')).toPandas()
song_avg
# plot the comparison of churned users and not by the next song page
sns.barplot(x='churn_group', y='count', data = song_churn)
plt.title('Number of Users by Total Number of Songs Played')
plt.xlabel('Churned or Not')
plt.ylabel('Song')
plt.show()
# plot the comparison of churned users and not by the next song page
sns.barplot(x='churn_group', y='avgsong', data = song_avg)
plt.title('Avg number of Users by Total Number of Songs Played')
plt.xlabel('Churned or Not')
plt.ylabel('Song')
plt.show()<jupyter_output><empty_output><jupyter_text>From the graph above, churned users have played way less songs than the non-churned one.  Average Number of Songs Played per Day for Churned and Non-Churned Users<jupyter_code># filter to the next song action page only
# convert timestamp(ts) to day format
song_day =  df_clean.where('page == "NextSong"').withColumn('ts', F.date_trunc('day', F.col('ts')))
# aggregate the number of page per day group by user id and the day 
song_day = song_day.groupBy(['userId', 'ts']).agg(F.count(F.col('page')).alias('numSong'))
# group by again by users and get the avg of num of page next song per day for the user
song_day = song_day. groupBy('userId').agg(F.avg(F.col('numSong')).alias('AvgNumSongPerDay'))
# join the table with the df churn table and convert to Pandas
pd_songday_churned = song_day.join(df_churn, ['userId']).toPandas()
pd_songday_churned.head()
# descriptive stat for avg song played per day by churn group
pd_songday_churned.groupby('churn_group').describe()<jupyter_output><empty_output><jupyter_text>Based on the descriptive stat, I would like to visualize both the non-churned and churned distribution. Although the mean is quite close, the median for churned group is lower than the non-churned one<jupyter_code># Sort the dataframe based on churn and not churn
avgsong_nochurn = pd_songday_churned.loc[pd_songday_churned['churn_group'] == 0]
avgsong_churn = pd_songday_churned.loc[pd_songday_churned['churn_group'] == 1]

# plot the distribution of churn and not churn users
sns.distplot(avgsong_nochurn[['AvgNumSongPerDay']], label = 'not_churn', color = 'green')
sns.distplot(avgsong_churn[['AvgNumSongPerDay']], label = 'churn', color = 'orange')
plt.title('Average Number of Song per Day')
plt.xlabel('Average Song per Day')
plt.ylabel('User Density')

plt.legend()
plt.show()<jupyter_output><empty_output><jupyter_text>Number of Songs Played per Hour<jupyter_code># filter to the next song action page only
# convert timestamp(ts) to hour format
song_hour =  df_clean.where('page == "NextSong"').withColumn('hour_ts', F.date_trunc('hour', F.col('ts')))
# aggregate the number of page per day group by user id and the day 
song_hour = song_hour.select(['song', 'hour_ts']).groupBy(['hour_ts']).agg(F.count(F.col('song')).alias('song')).orderBy('hour_ts').alias('countSong').toPandas()
song_hour.head()<jupyter_output><empty_output><jupyter_text>Length of Song per Day for Churned and Non-Churned Users<jupyter_code># filter to the next song action page only
# convert timestamp(ts) to day format
# length is defined for length of song (in second) played since it occurs only at page nextsong
length_song_day =  df_clean.withColumn('ts', F.date_trunc('day', F.col('ts')))
# aggregate the number of length per day group by user id and the day 
length_song_day = length_song_day.groupBy(['userId', 'ts']).agg(F.sum(F.col('length')).alias('lengthSongPerDay'))
# group by again by users and get the avg of num of length song per day for the user
length_song_day = length_song_day.groupBy('userId').agg(F.avg(F.col('lengthSongPerDay')).alias('AvgLengthSongPerDay'))
# join the table with the df churn table and convert to Pandas
pd_lengthsongday_churned = length_song_day.join(df_churn, ['userId']).toPandas()
pd_lengthsongday_churned.head()
# descriptive stat for avg length song played per day by churn group
pd_lengthsongday_churned.groupby('churn_group').describe()<jupyter_output><empty_output><jupyter_text>Based on the descriptive stat, I would like to visualize both the non-churned and churned distribution. The median for churned group is lower than the non-churned one as well as the mean<jupyter_code># Sort the dataframe based on churn and not churn
avglength_song_nochurn = pd_lengthsongday_churned.loc[pd_lengthsongday_churned['churn_group'] == 0]
avglength_song_churn = pd_lengthsongday_churned.loc[pd_lengthsongday_churned['churn_group'] == 1]

# plot the distribution of churn and not churn users
sns.distplot(avglength_song_nochurn[['AvgLengthSongPerDay']], label = 'not_churn', color = 'green')
sns.distplot(avglength_song_churn[['AvgLengthSongPerDay']], label = 'churn', color = 'orange')
plt.title('Average Length of Song per Day')
plt.xlabel('Average Length per Day')
plt.ylabel('User Density')

plt.legend()
plt.show()<jupyter_output><empty_output><jupyter_text>Number of Item in Sessions for Churned and Non-Churned Users<jupyter_code># user id and session id is not unique, we use max value for iteminsession, since it increases incrementally in 1 session id
item_churned =  df_clean.select('userId', 'sessionId', 'ItemInSession').groupBy('userId', 'sessionId').agg(F.max(F.col('itemInSession')).alias('totalItem'))
item_churned.show()
item_churned = item_churned.groupBy('userId').agg(F.avg(F.col('totalItem')).alias('avgItemInSession'))
item_churned.show()
pd_item_churned.head()
# descriptive stat for no of item in session  per day by churn group
pd_item_churned.groupby('churn_group').describe()<jupyter_output><empty_output><jupyter_text>Based on the descriptive stat, I would like to visualize both the non-churned and churned distribution. The median for churned group is significantly lower than the non-churned one as well as the mean.<jupyter_code># join the item per session table with the churn df and convert to Pandas
pd_item_churned = item_churned.join(df_churn, ['userId']).toPandas()

# Sort the dataframe based on churn and not churn
itemsess_nochurn = pd_item_churned.loc[pd_item_churned['churn_group'] == 0]
itemsess_churn = pd_item_churned.loc[pd_item_churned['churn_group'] == 1]

# plot the distribution of churn and not churn users
sns.distplot(itemsess_nochurn[['avgItemInSession']], label = 'no_churn', color = 'green')
sns.distplot(itemsess_churn[['avgItemInSession']], label = 'churn', color ='orange')
plt.title('Average Activities per Session')
plt.xlabel('Average Number of Items per Session')
plt.ylabel('Total User')

plt.legend()
plt.show()<jupyter_output><empty_output><jupyter_text>From the distribution plot above, the churn distributor is more rightly skewed that the no churn users.  Average number of session per day for those who churned and not<jupyter_code># convert timestamp(ts) to day format
item_session_day =  df_clean.withColumn('ts', F.date_trunc('day', F.col('ts')))
# aggregate the number of session per day group by user id and the day 
item_session_day = item_session_day.groupBy(['userId', 'ts']).agg(F.countDistinct(F.col('sessionId')).alias('numSession'))
# group by again by users and get the avg of num of session per day for the user
item_session_day = item_session_day.groupBy('userId').agg(F.avg(F.col('numSession')).alias('AvgNumSessionPerDay'))
# join the table with the df churn table and convert to Pandas
pd_sessday_churned = item_session_day.join(df_churn, ['userId']).toPandas()
pd_sessday_churned.head()
# descriptive stat for no of session  per day by churn group
pd_sessday_churned.groupby('churn_group').describe()<jupyter_output><empty_output><jupyter_text>Based on the descriptive stat, I would like to visualize both the non-churned and churned distribution. The median for churned group is higher than the non-churned one as well as the mean<jupyter_code># Sort the dataframe based on churn and not churn
avgsess_nochurn = pd_sessday_churned.loc[pd_sessday_churned['churn_group'] == 0]
avgsess_churn = pd_sessday_churned.loc[pd_sessday_churned['churn_group'] == 1]

# plot the distribution of churn and not churn users
sns.distplot(avgsess_nochurn[['AvgNumSessionPerDay']], label = 'no_churn', color = 'green')
sns.distplot(avgsess_churn[['AvgNumSessionPerDay']], label = 'churn', color = 'orange')
plt.title('Average Number of Session per Day')
plt.xlabel('Average Session per Day')
plt.ylabel('User Density')

plt.legend()
plt.show()<jupyter_output><empty_output><jupyter_text>From the distribution plot above, the churned users have lower maximum value and have 'fatter' distribution compared to the non-churned one.   Thumbs Down and Thumbs Up Users<jupyter_code>td_churn = df_clean.where(df_clean['page'] == 'Thumbs Down').groupby(["churn_group"]).count().toPandas()
td_churn
# plot no of users who churned and not by thumbs down page
sns.barplot(x='churn_group', y='count', data = td_churn)
plt.title('Number of Users by Total Thumbs Down')
plt.xlabel('Churned or Not')
plt.ylabel('Total User')
plt.show()
tu_churn = df_clean.where(df_clean['page'] == 'Thumbs Up').groupby(["churn_group"]).count().toPandas()
tu_churn
# plot no of users who churned and not by thumbs ip page
sns.barplot(x='churn_group', y='count', data = tu_churn)
plt.title('Number of Users by Total Thumbs Up')
plt.xlabel('Churned or Not')
plt.ylabel('Total User')
plt.show()<jupyter_output><empty_output><jupyter_text>Feature EngineeringOnce you've familiarized yourself with the data, build out the features you find promising to train your model on. To work with the full dataset, you can follow the following steps.- Write a script to extract the necessary features from the smaller subset of data- Ensure that your script is scalable, using the best practices discussed in Lesson 3- Try your script on the full data set, debugging your script if necessaryIf you are working in the classroom workspace, you can just extract features based on the small subset of data contained here. Be sure to transfer over this work to the larger dataset when you work on your Spark cluster.<jupyter_code># check the existing df
df_churn.show()
# check the existing df
df_clean.printSchema()<jupyter_output>root
 |-- userId: string (nullable = true)
 |-- artist: string (nullable = true)
 |-- auth: string (nullable = true)
 |-- firstName: string (nullable = true)
 |-- gender: string (nullable = true)
 |-- itemInSession: long (nullable = true)
 |-- lastName: string (nullable = true)
 |-- length: double (nullable = true)
 |-- level: string (nullable = true)
 |-- location: string (nullable = true)
 |-- method: string (nullable = true)
 |-- page: string (nullable = true)
 |-- registration: string (nullable = true)
 |-- sessionId: long (nullable = true)
 |-- song: string (nullable = true)
 |-- status: long (nullable = true)
 |-- ts: string (nullable = true)
 |-- userAgent: string (nullable = true)
 |-- churn_group: long (nullable = true)<jupyter_text>Extracting Features<jupyter_code># new df for model training
df_model = df_churn<jupyter_output><empty_output><jupyter_text>Feature from Page Column<jupyter_code># selecting all pages except for nextsong and cancellation confirmation  and make it into list
pages = df_clean.select('page').distinct().toPandas()
list_page = pages.page.tolist()

list_page.remove('Cancellation Confirmation')
list_page.remove('NextSong')
list_page
# create a function to extract features from the page columns
def page_feature(df, page_name):
    page_df = df.where(df.page == page_name).groupBy(['userId']).\
    agg(F.count(F.col('page')).alias('Count' + page_name.replace(' ', ''))).drop('page')
    return page_df
# loop through the name of pages and use the function
# join the new df with the df where pages is extracted
for feature in list_page:
    page_df = page_feature(df_clean, feature)
    df_model = df_model.join(page_df, ['userId'], 'left').fillna(0)
df_model.head()<jupyter_output><empty_output><jupyter_text>Feature from Gender Column<jupyter_code># selecting the gender column
gender_df = df_clean.select('userId', 'gender').distinct()
# converting the gender into integer type with lambda function
gender_categories = F.udf(lambda gender: 0 if gender == 'M' else 1, IntegerType())
# adding gender categories as a new column
gender_df = gender_df.withColumn('gender', gender_categories(F.col('gender')))
# joining gender column to the model colum
df_model = df_model.join(gender_df, ['userId'])
# checking the new gender column
df_model.head()<jupyter_output><empty_output><jupyter_text>Feature from Level Column - Paid or Free Users?<jupyter_code># selecting the level column
level_df = df_clean.select('userId', 'level').groupBy('userId').agg(F.max(F.col('level')).alias('isPaid'))
# converting the level into integer type with lambda funciton
level_category = F.udf(lambda level: 0 if level == 'free' else 1, IntegerType())
# adding level category as a new column
level_df = level_df.withColumn('isPaid', level_category(F.col('isPaid')))
# joining level column to the model colum
df_model = df_model.join(level_df, ['userId'])
# checking the new ispaid column
df_model.head()<jupyter_output><empty_output><jupyter_text>Feature From Page Column - Total Number of Song Played by User<jupyter_code># selecting user id and page and filter it only on the nextsong and aggregate it to find the count
song_df = df_clean.select('userId', 'page').where(df_clean['page'] == 'NextSong').groupby(['userId']).agg(F.count(F.col('page')).alias('countSong'))

# joining song df to the model df
df_model = df_model.join(song_df, ['userId'])                                                                                                      
# checking the count song column
df_model.head()<jupyter_output><empty_output><jupyter_text>Feature From IteminSession Column - Average Number of Item in 1 Session<jupyter_code># selectng user session id and item in session
avgItemSession = df_clean.select('userId', 'sessionId', 'ItemInSession')
# group by user and session and find the number of item per session
avgItemSession = avgItemSession.groupBy('userId', 'sessionId').agg(F.max(F.col('itemInSession')).alias('numItem'))
# find the average
avgItemSession = avgItemSession.groupBy('userId').agg(F.avg(F.col('numItem')).alias('avgNumItemSession'))

# joining the avg item session table to model df
df_model = df_model.join(avgItemSession, ['userId'])
df_model.head()<jupyter_output><empty_output><jupyter_text>Feature From SessionId Column - Average Number of Session per Day<jupyter_code># convert timestamp(ts) to day format
sessDay_df =  df_clean.withColumn('ts', F.date_trunc('day', F.col('ts')))
# aggregate the number of session per day group by user id and the day 
sessDay_df = sessDay_df.groupBy(['userId', 'ts']).agg(F.countDistinct(F.col('sessionId')).alias('numSession'))
# group by again by users and get the avg of num of session per day for the user
sessDay_df = sessDay_df.groupBy('userId').agg(F.avg(F.col('numSession')).alias('AvgNumSessionPerDay'))
# join the table with the df churn table and convert to Pandas
df_model = df_model.join(sessDay_df, ['userId'])
df_model.head()<jupyter_output><empty_output><jupyter_text>Feature From Page NextSong Column - Average Number of Song Played per Day<jupyter_code># filter to the next song action page only
# convert timestamp(ts) to day format
avgSong_day =  df_clean.where('page == "NextSong"').withColumn('ts', F.date_trunc('day', F.col('ts')))
# aggregate the number of page per day group by user id and the day 
avgSong_day = avgSong_day.groupBy(['userId', 'ts']).agg(F.count(F.col('page')).alias('numSong'))
# group by again by users and get the avg of num of page next song per day for the user
avgSong_day = avgSong_day.groupBy('userId').agg(F.avg(F.col('numSong')).alias('AvgNumSongPerDay'))
# join the table with the df churn table and convert to Pandas
df_model = df_model.join(avgSong_day, ['userId'])
# check avg song per day
df_model.head()<jupyter_output><empty_output><jupyter_text>Feature From Length Column - Average Length of Song Played per Day<jupyter_code># convert timestamp(ts) to day format
# length is defined for length of song (in second) played since it occurs only at page nextsong
avgLengthsong_day =  df_clean.withColumn('ts', F.date_trunc('day', F.col('ts')))
# aggregate the number of length per day group by user id and the day 
avgLengthsong_day = avgLengthsong_day.groupBy(['userId', 'ts']).agg(F.sum(F.col('length')).alias('lengthSongPerDay'))
# group by again by users and get the avg of num of length song per day for the user
avgLengthsong_day = avgLengthsong_day.groupBy('userId').agg(F.avg(F.col('lengthSongPerDay')).alias('AvgLengthSongPerDay'))
# join the table with the df churn table and convert to Pandas
df_model = df_model.join(avgLengthsong_day, ['userId'])
# check avg length per day
df_model.head()<jupyter_output><empty_output><jupyter_text>Check Features<jupyter_code>df_model.count()
df_model.printSchema()
#count null value from each column
for column in df_model.columns:
    print(column, "missing values:", df_model.filter(F.isnull(df_model[column])).count())<jupyter_output>userId missing values: 0
churn_group missing values: 0
CountCancel missing values: 0
CountSubmitDowngrade missing values: 0
CountThumbsDown missing values: 0
CountHome missing values: 0
CountDowngrade missing values: 0
CountRollAdvert missing values: 0
CountLogout missing values: 0
CountSaveSettings missing values: 0
CountAbout missing values: 0
CountSettings missing values: 0
CountAddtoPlaylist missing values: 0
CountAddFriend missing values: 0
CountThumbsUp missing values: 0
CountHelp missing values: 0
CountUpgrade missing values: 0
CountError missing values: 0
CountSubmitUpgrade missing values: 0
gender missing values: 0
isPaid missing values: 0
countSong missing values: 0
avgNumItemSession missing values: 0
AvgNumSessionPerDay missing values: 0
AvgNumSongPerDay missing values: 0
AvgLengthSongPerDay missing values: 0<jupyter_text>Vectorize<jupyter_code># create vector assembler
features = df_model.columns

# removing other than the fetures for the list
features.remove('churn_group')
features.remove('userId')
features
# create the vector assembler and use it to vectorize the feature
assembler = VectorAssembler(inputCols=features, outputCol='vectorizedFeatures')
df_feature_assemble = assembler.transform(df_model)
df_feature_assemble.head()<jupyter_output><empty_output><jupyter_text>Scale Features<jupyter_code># scale the features 
scaler = Normalizer(inputCol = 'vectorizedFeatures', outputCol = 'scaled_features')
df_scale = scaler.transform(df_feature_assemble)
#df_scale.head()
# the data created which consist of lable and (vectorized and scaled) features will be used for modeling, hence we need to label it 
df_ml_model = df_scale.select(F.col('churn_group').alias('label'), F.col('scaled_features').alias('features'))
df_ml_model.head()
df_ml_model.printSchema()<jupyter_output>root
 |-- label: long (nullable = true)
 |-- features: vector (nullable = true)<jupyter_text>ModelingSplit the full dataset into train, test, and validation sets. Test out several of the machine learning methods you learned. Evaluate the accuracy of the various models, tuning parameters as necessary. Determine your winning model based on test accuracy and report results on the validation set. Since the churned users are a fairly small subset, I suggest using F1 score as the metric to optimize.  Train and Test Data<jupyter_code># break the dataset into training and testing sets
train, test = df_ml_model.randomSplit([.8, .2], seed=42)<jupyter_output><empty_output><jupyter_text>Random Forest Classifier<jupyter_code>evaluator = MulticlassClassificationEvaluator(metricName = 'f1')
# define the model
rf = RandomForestClassifier()

# define the parameter alternatives to be tested
rf_paramGrid = ParamGridBuilder() \
            .addGrid(rf.numTrees,[5, 10]) \
            .addGrid(rf.maxDepth,[2, 5]) \
            .build()

# define the cross validator
rf_cross_validation1 = CrossValidator(estimator = rf,
                          estimatorParamMaps = rf_paramGrid,
                          evaluator = evaluator,
                          numFolds =3)


# train the model
start = time.time()
rf_model1 = rf_cross_validation1.fit(train)
end = time.time()
print(f'Model training is done, spent {end-start}s.')
rf_model1.avgMetrics
# make prediction with the model
rf_pred1 = rf_model1.transform(test)
# calculate the F1 score
rf_f1_score = evaluator.evaluate(rf_pred1)
print('Random Forest F1-Score: {}'.format(round(rf_f1_score)))

# calculate the accuracy, ref calculation from spark lecture
rf_accuracy = rf_pred1.filter(rf_pred1.label == rf_pred1.prediction).count() / rf_pred1.count() 
print('Random Forest Accuracy:', rf_accuracy)<jupyter_output>Random Forest F1-Score: 1
Random Forest Accuracy: 1.0<jupyter_text>Logistic Regression<jupyter_code># define the model
lr1 = LogisticRegression()

# define the parameter alternatives to be tested
lr_paramGrid1 = ParamGridBuilder()\
                .addGrid(lr1.regParam, [0.0, 0.1])\
                .addGrid(lr1.maxIter, [5, 10])\
                .build()

# define the cross validator
lr_cross_validation1 = CrossValidator(estimator=lr1,
                              estimatorParamMaps=lr_paramGrid1,
                              evaluator = evaluator,
                              numFolds=3)

# train the model
start = time.time()
lr_model1 = lr_cross_validation1.fit(train)
end = time.time()
print(f'Model training is done, spent {end-start}s.')
lr_model1.avgMetrics
# make prediction with the model
lr_pred1 = lr_model1.transform(test) 
# calculate the accuracy, ref calculation from spark lecture
accuracy_lr1 = lr_pred1.filter(lr_pred1.label == lr_pred1.prediction).count() / lr_pred1.count()
print('Logistic Regression Accuracy:', accuracy_lr1)
# calculate the F1 score
lr_f1_score = evaluator.evaluate(lr_pred1)
print('Logistic Regresson F1-Score: {}'.format((lr_f1_score)))<jupyter_output>Logistic Regresson F1-Score: 0.9701143615340873<jupyter_text>Gradient Boosted Tree<jupyter_code># define the model
gbt1 = GBTClassifier()

# define the parameter alternatives to be tested
gbt_paramGrid1 = ParamGridBuilder()\
                .addGrid(gbt1.maxIter, [5, 10])\
                .addGrid(gbt1.maxDepth, [2, 5])\
                .build()

# define the cross validator
gbt_cross_validation1 = CrossValidator(estimator=gbt1,
                              estimatorParamMaps=gbt_paramGrid1,
                              evaluator = evaluator,
                              numFolds=3)

# train the model
start = time.time()
gbt_model1 = gbt_cross_validation1.fit(train)
end = time.time()
print(f'Model training is done, spent {end-start}s.')
gbt_model1.avgMetrics
# make prediction with the model
gbt_pred1 = gbt_model1.transform(test)
# calculate the accuracy, ref calculation from spark lecture
accuracy_gbt1 = gbt_pred1.filter(gbt_pred1.label == gbt_pred1.prediction).count() / gbt_pred1.count()
print('Gradient Boosted Accuracy:', accuracy_gbt1)
# calculate the F1 score
gbt_f1_score = evaluator.evaluate(gbt_pred1)
print('Gradient Boosted F1-Score: {}'.format((gbt_f1_score)))<jupyter_output>Gradient Boosted F1-Score: 1.0<jupyter_text>Naive Bayes<jupyter_code># define the model
nb1 = NaiveBayes()

# define the parameter alternatives to be tested
nb_paramGrid1 = ParamGridBuilder()\
                .addGrid(nb1.smoothing, [0.0, 0.5])\
                .build()

# define the cross validator
nb_cross_validation1 = CrossValidator(estimator=nb1,
                              estimatorParamMaps=nb_paramGrid1,
                              evaluator=evaluator,
                              numFolds=3)

# train the model
start = time.time()
nb_model1 = nb_cross_validation1.fit(train)
end = time.time()
print(f'Model training is done, spent {end-start}s.')
nb_model1.avgMetrics
# make prediction with the model
nb_pred1 = nb_model1.transform(test)
accuracy_nb1 = nb_pred1.filter(nb_pred1.label == nb_pred1.prediction).count() / nb_pred1.count()
print('Naive Bayes Accuracy:', accuracy_nb1)
# calculate the F1 score
nb_f1_score = evaluator.evaluate(nb_pred1)
print('Gradient Boosted F1-Score: {}'.format((nb_f1_score)))<jupyter_output>Gradient Boosted F1-Score: 1.0<jupyter_text>Linear Support Vector Machine<jupyter_code># define the model
lsvm1 = LinearSVC()

# define the parameter alternatives to be tested
lsvm_paramGrid1 = ParamGridBuilder()\
                .addGrid(lsvm1.maxIter, [2, 5])\
                .addGrid(lsvm1.regParam, [0.0, 0.1])\
                .build()

# define the cross validator
lsvm_cross_validation1 = CrossValidator(estimator=lsvm1,
                              estimatorParamMaps=lsvm_paramGrid1,
                              evaluator = evaluator,
                              numFolds=3)

# train the model
start = time.time()
lsvm_model1 = lsvm_cross_validation1.fit(train)
end = time.time()
print(f'Model t is done, spent {end-start}s.')
lsvm_model1.avgMetrics
lsvm_pred1 = lsvm_model1.transform(test)
# calculate the accuracy, ref calculation from spark lecture
accuracy_lsvm1 = lsvm_pred1.filter(lsvm_pred1.label == lsvm_pred1.prediction).count() / lsvm_pred1.count()
print('Linear Support Vector Machine Accuracy:', accuracy_lsvm1)
# calculate the F1 score
lsvm_f1_score = evaluator.evaluate(lsvm_pred1)
print('Linear Support Vector Machine F1-Score: {}'.format((lsvm_f1_score)))<jupyter_output>Linear Support Vector Machine F1-Score: 0.5841784989858012
{'fraction_non_alphanumeric': 0.1275276354812618, 'fraction_numerical': 0.029069339934802322, 'mean_word_length': 4.175694532538373, 'pattern_counts': {'":': 0, '<': 170, '<?xml version=': 0, '>': 170, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '46', 'dir': 'jupyter-structured-clean-dedup', 'id': '324686', 'license': 'MIT', 'n_tokens_mistral': 13990, 'n_tokens_neox': 12961, 'n_words': 3870, 'path': 'sparkify.ipynb', 'repo_name': 'WidyaPuspitaloka/sparkify_project', 'size': '40826'}