<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.Link to Github repo: https://github.com/martinkirov/dsnd-capstone Link to blogpost on Medium: https://medium.com/@martin701993/sparkify-capstone-project-86c9618d9c95<jupyter_code># import libraries
import pyspark
from pyspark import SparkConf
from pyspark.sql import SparkSession
from pyspark.sql.functions import udf
import datetime
from pyspark.sql.types import IntegerType
from pyspark.sql import Window

import seaborn as sns
from matplotlib import pyplot as plt
import pandas as pd

from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.feature import MinMaxScaler
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.classification import GBTClassifier

from pyspark.sql import Window
import pyspark.sql.functions as f

from pyspark.ml import Pipeline
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
# create a Spark session
spark = SparkSession \
    .builder \
    .appName("EDA using pyspark") \
    .getOrCreate()
spark<jupyter_output><empty_output><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>path = "mini_sparkify_event_data.json"
user_log = spark.read.json(path)
user_log.printSchema()
user_log.take(3)
# Total number of rows
user_log.count()
# Find the blanks in non-string columns
["{}: {}".format(col,user_log.filter(user_log[col].isNull()).count()) for col in user_log.columns]
# Get the blanks in string columns
["{}: {}".format(col,user_log.filter(user_log[col] == "").count()) for col in user_log.columns]<jupyter_output><empty_output><jupyter_text>MK Comment:From the given columns, we need to check out:+ userId (8k)+ userAgent (8k)+ firstName (8k)+ gender (8k)+ lastName (8k)+ location (8k)+ registration (8k)+ artist (58k)+ length (58k)+ song (58k)Lets check if the columns with 8k missing records match up:<jupyter_code>user_log \
.select(['userId', 'userAgent', 'firstName', 'gender', 'lastName', 'location', 'registration', 'artist']) \
.filter(user_log['userAgent'].isNull() &  user_log['gender'].isNotNull() ) \
.show()<jupyter_output>+------+---------+---------+------+--------+--------+------------+------+
|userId|userAgent|firstName|gender|lastName|location|registration|artist|
+------+---------+---------+------+--------+--------+------------+------+
+------+---------+---------+------+--------+--------+------------+------+<jupyter_text>MK Comment:The 8k features do match up; these are all logged out screens which we arent too interested in and we'll drop.How about the features with 58k missing values?<jupyter_code>user_log \
.filter(user_log.artist.isNull()) \
.select('userId','method', 'page','status','sessionId','registration') \
.show()
user_log \
.filter(user_log.sessionId == 187) \
.sort('ts') \
.select('ts', 'userAgent','method', 'page','status','sessionId','registration') \
.show(1000)<jupyter_output>+-------------+--------------------+------+---------------+------+---------+-------------+
|           ts|           userAgent|method|           page|status|sessionId| registration|
+-------------+--------------------+------+---------------+------+---------+-------------+
|1538359181000|"Mozilla/5.0 (Win...|   GET|           Home|   200|      187|1538304455000|
|1538359204000|"Mozilla/5.0 (Win...|   PUT|       NextSong|   200|      187|1538304455000|
|1538359398000|"Mozilla/5.0 (Win...|   PUT|       NextSong|   200|      187|1538304455000|
|1538359596000|"Mozilla/5.0 (Win...|   PUT|       NextSong|   200|      187|1538304455000|
|1538359749000|"Mozilla/5.0 (Win...|   PUT|       NextSong|   200|      187|1538304455000|
|1538359971000|"Mozilla/5.0 (Win...|   PUT|       NextSong|   200|      187|1538304455000|
|1538360229000|"Mozilla/5.0 (Win...|   PUT|       NextSong|   200|      187|1538304455000|
|1538360490000|"Mozilla/5.0 (Win...|   PUT|       NextSong|   200|      187|1538304455000|[...]<jupyter_text>`registration` and `ts` need to be reformatted.<jupyter_code>convert_ts = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).strftime("%Y-%m-%d %H:%M:%S"))
user_log = user_log.withColumn('event_time', convert_ts('ts'))
user_log = user_log.withColumn('registration_time', convert_ts('registration'))
log_data = user_log.filter(user_log['userId'] != "")<jupyter_output><empty_output><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># define the churn and downgrade events we want to flag
churn_flag = udf(lambda x: 1 if x == 'Cancellation Confirmation' else 0, IntegerType())
downgrade_flag = udf(lambda x: 1 if x == 'Submit Downgrade' else 0, IntegerType())
# create churn_event and downgrade_event columns
log_data = log_data.withColumn('churn_event', churn_flag("page"))
log_data = log_data.withColumn('downgrade_event', downgrade_flag("page"))
# label users that have ever churned
churned_users = set(log_data.filter(log_data['churn_event'] == 1).select('userId').dropDuplicates().toPandas().userId)
churned_user = udf(lambda x: 1 if x in churned_users else 0, IntegerType())
log_data = log_data.withColumn('churn_user', churned_user('userId'))
# label users that have ever downgraded
downgraded_users = set(log_data.filter(log_data['downgrade_event'] == 1).select('userId').dropDuplicates().toPandas().userId)
downgrade_user = udf(lambda x: 1 if x in downgraded_users else 0, IntegerType())
log_data = log_data.withColumn('downgrade_user', downgrade_user('userId'))
df = log_data.toPandas()<jupyter_output><empty_output><jupyter_text>2. Exploratory Analysis 2.1 High level planWe will treat the EDA part as a learning experience to learn about the dataset and get enough insight so as to construct a useful end-table that we will apply ML algorithms on.  2.2. Basic Info<jupyter_code>df.shape
df.head().transpose()
df.info()
df.describe().transpose()<jupyter_output><empty_output><jupyter_text>How many unique values do we have for each feature? (both categorical and continuous)Note: we will gather all the variables that have less than 100 unique values into `cols_to_check_further` to explore further.<jupyter_code>## Unique vals in each var
cols_to_check_further = []
for col in df.columns:
    no_unique_vals = len(df[col].unique())
    print("{}: {}".format(col, no_unique_vals))
    if no_unique_vals < 100:
        cols_to_check_further.append(col)<jupyter_output>artist: 17656
auth: 2
firstName: 189
gender: 2
itemInSession: 1311
lastName: 173
length: 14866
level: 2
location: 114
method: 2
page: 19
registration: 225
sessionId: 2312
song: 58481
status: 3
ts: 269770
userAgent: 56
userId: 225
event_time: 269770
registration_time: 225
churn_event: 2
downgrade_event: 2
churn_user: 2
downgrade_user: 2<jupyter_text>2.2.1. Unique Value CountsFor ease of analysis, lets construct a dictionary, `uvc`, that will hold all the value counts for the variables in `cols_to_check_further`.<jupyter_code># uncomment for activity-based counting vs user-based counting
"""
uvc = {}
for col in cols_to_check_further:
    uvc[col] = pd.DataFrame(df[col].value_counts())
"""
uvc = {}
for col in cols_to_check_further:
    uvc[col] = pd.DataFrame(df[['userId', col]].drop_duplicates()[col].value_counts())
cols_to_check_further
uvc['auth']
uvc['gender']
uvc['method']
uvc['level']
uvc['userAgent'].head()
# Get the number of users that have moved between paid and free (both directions)

free_paid_users = df[['userId', 'level']].drop_duplicates()

len(free_paid_users[free_paid_users.duplicated(subset="userId")].sort_values(by="userId"))<jupyter_output><empty_output><jupyter_text>2.3. Single Var ExplorationThe above dataframes give us the exact value counts within each variable but it not too convenient when it comes to getting a feel for the data. Let's chart out all of the `uvc` data and see - in percentage terms - how they are distributed.<jupyter_code>for col in uvc.keys():
    uvc[col].plot(kind='bar', legend=True, title=col, figsize = (12,4))

    #reformat the y-axis to percentages
    plt.gca().set_yticklabels(['{:.0f}%'.format(100*x/225) for x in plt.gca().get_yticks()])
    plt.show()<jupyter_output><empty_output><jupyter_text>2.4. Non-Categorical variable (sanity) check<jupyter_code>nc_vars = list(set(df.columns) - set(uvc.keys()) - set(['userId']))
user = '126'
df[df['userId'] == user][nc_vars]<jupyter_output><empty_output><jupyter_text>2.5. Number of listening sessions on SparkifyLet us consider the number of listening sessions that Sparkify users have. From the sample data we have, the `sessionId` feature gives the unique id of the session that the user is currently listening to. Every single user will have one or more sessions. We would like to see how the users are distributed based on the sessions they have.<jupyter_code>xx = df.groupby(['userId'])
xx.get_group('125').sessionId
no_sessions = xx.sessionId.nunique().sort_values(ascending = False)
pd.cut(no_sessions, 20).value_counts(sort=False).plot(kind='barh', \
                                                      title = "Number of listening sessions on Sparkify", \
                                                      figsize = (10,8));<jupyter_output><empty_output><jupyter_text>That's interesting - about 70 users or ~ 30% of the users in our dataset only use Sparkify a few times (between 0 and 6 times), and then the more sessions we consider, the less users have reach the number as a milestone.  2.6. What devices do users haveLet's consider what information we can obtain from the `userAgent` variable. There are about 55 different types of userAgents and it's not possible to figure out what browser and machine the user had - at least not without external help. Luckily for us, there are websites (I used this one: https://developers.whatismybrowser.com/useragents/parse/) where you can input a `userAgent` signature and you'll get back the browser and machine used. It took me about 15 mins to manually enter each of the 55 and parse the data out -- the `test` variable below contains what I parsed in the website and what we will be joining to our data later on.<jupyter_code>userAgents = pd.DataFrame(df.userAgent.value_counts())
# this is manually parsed browser and machine used for each UserAgent
test = """userAgent	Browser	Machine	
\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Windows 7	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.77.4 (KHTML, like Gecko) Version/7.0.5 Safari/537.77.4\"	Safari 7	Mavericks	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Mavericks	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Mavericks	
Mozilla/5.0 (Windows NT 6.1; WOW64; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Windows 7	
\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Windows 7	
\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/37.0.2062.103 Safari/537.36\"	Chrome 37	Windows 7	
Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Mavericks	
\"Mozilla/5.0 (iPad; CPU OS 7_1_2 like Mac OS X) AppleWebKit/537.51.2 (KHTML, like Gecko) Version/7.0 Mobile/11D257 Safari/9537.53\"	Safari 7	iPad	
Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; WOW64; Trident/5.0)	IE 9	Windows 7	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.78.2 (KHTML, like Gecko) Version/7.0.6 Safari/537.78.2\"	Safari 7	Mavericks	
\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Windows 7	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/37.0.2062.94 Safari/537.36\"	Chrome 37	Mavericks	
\"Mozilla/5.0 (Windows NT 6.3; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Windows 8_1	
\"Mozilla/5.0 (iPhone; CPU iPhone OS 7_1_2 like Mac OS X) AppleWebKit/537.51.2 (KHTML, like Gecko) Version/7.0 Mobile/11D257 Safari/9537.53\"	Safari 7	iPhone	
Mozilla/5.0 (Windows NT 6.3; WOW64; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Windows 8.1	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10) AppleWebKit/600.1.8 (KHTML, like Gecko) Version/8.0 Safari/600.1.8\"	Safari 8	Yosemite	
\"Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Windows XP	
\"Mozilla/5.0 (Windows NT 6.3; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Windows 8.1	
Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0; rv:11.0) like Gecko	IE 11	Windows 7	
\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/37.0.2062.94 Safari/537.36\"	Chrome 37	Windows 7	
Mozilla/5.0 (Windows NT 6.1; WOW64; rv:32.0) Gecko/20100101 Firefox/32.0	Firefox 32	Windows 7	
Mozilla/5.0 (Windows NT 6.1; WOW64; rv:30.0) Gecko/20100101 Firefox/30.0	Firefox 30	Windows 7	
\"Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Windows 8	
\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrofme 36	Linux	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_2) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Mavericks	
\"Mozilla/5.0 (iPhone; CPU iPhone OS 7_1 like Mac OS X) AppleWebKit/537.51.2 (KHTML, like Gecko) Version/7.0 Mobile/11D167 Safari/9537.53\"	Safari 7	iPhone	
Mozilla/5.0 (X11; Linux x86_64; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Linux	
\"Mozilla/5.0 (iPad; CPU OS 7_1_1 like Mac OS X) AppleWebKit/537.51.2 (KHTML, like Gecko) Version/7.0 Mobile/11D201 Safari/9537.53\"	Safari 7	iPad	
Mozilla/5.0 (Macintosh; Intel Mac OS X 10.7; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Lion	
Mozilla/5.0 (Windows NT 6.0; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	WIndows Vista	
\"Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.153 Safari/537.36\"	Chrome 35	Windows 7	
\"Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Windows XP	
Mozilla/5.0 (Windows NT 6.2; WOW64; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Windows 8	
\"Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Windows 7	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.76.4 (KHTML, like Gecko) Version/7.0.4 Safari/537.76.4\"	Safari 7	Mavericks	
\"Mozilla/5.0 (iPhone; CPU iPhone OS 7_1_1 like Mac OS X) AppleWebKit/537.51.2 (KHTML, like Gecko) Version/7.0 Mobile/11D201 Safari/9537.53\"	Safari 7	iPhone	
Mozilla/5.0 (Macintosh; Intel Mac OS X 10.8; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Mountain Lion	
Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Linux	
\"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Linux	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_10) AppleWebKit/600.1.3 (KHTML, like Gecko) Version/8.0 Safari/600.1.3\"	Safari 8	Yosemite	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_4) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.153 Safari/537.36\"	Chrome 35	Mavericks	
Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)	IE 9	Windows 7	
Mozilla/5.0 (Windows NT 6.1; WOW64; rv:24.0) Gecko/20100101 Firefox/24.0	Firefox 24	Windows 7	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_2) AppleWebKit/537.74.9 (KHTML, like Gecko) Version/7.0.2 Safari/537.74.9\"	Safari 7	Mavericks	
Mozilla/5.0 (Windows NT 6.1; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Windows 7	
Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.1; WOW64; Trident/6.0)	IE 10	Windows 7	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_5) AppleWebKit/537.77.4 (KHTML, like Gecko) Version/6.1.5 Safari/537.77.4\"	Safari 6_1	Lion	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_2) AppleWebKit/537.75.14 (KHTML, like Gecko) Version/7.0.3 Safari/537.75.14\"	Safari 7	Mavericks	
\"Mozilla/5.0 (Windows NT 6.2; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Windows 8	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_7_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.125 Safari/537.36\"	Chrome 36	Lion	
Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Snow Leopard	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Mountain Lion	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_8_5) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/37.0.2062.94 Safari/537.36\"	Chrome 36	Mountain Lion	
\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_6_8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36\"	Chrome 36	Snow Leopard	
Mozilla/5.0 (X11; Ubuntu; Linux i686; rv:31.0) Gecko/20100101 Firefox/31.0	Firefox 31	Linux	"""
# create a 1-column dataframe with the manually-parsed data
x = pd.DataFrame(test.split('\t'))
# split the dataframe into three 1-col dataframes

# by userAgent
x0 = x[x.index % 3 == 0].reset_index()
# by browser
x1 = x[x.index % 3 == 1].reset_index()
# by machine used
x2 = x[x.index % 3 == 2].reset_index()
# data manipulation to line everything up
x = x1.join(x2, lsuffix='a', rsuffix='b')
x = x.drop(columns = ['indexa', 'indexb'])
x = x.rename(columns={'0a': 'browser', '0b': 'machine'})
x = x.iloc[1:]
x = x.reset_index().drop(columns = ['index'])
userAgents = userAgents.reset_index().join(x)
x = userAgents.groupby(['machine'])

x\
.agg('sum')\
.sort_values(['userAgent'], ascending = False)\
.plot(kind='barh', figsize=(12,8));
x = userAgents.groupby(['browser'])

x\
.agg('sum')\
.sort_values(['userAgent'], ascending = False)\
.plot(kind='barh', figsize=(12,8));<jupyter_output><empty_output><jupyter_text>So, from the above two charts, we can see that most of our users use desktops as machines and a large proportion of them are Windows-based (either Windows XP, 7, 8 or 8.1). In terms of browsers used, Chrome (ver 36) is the most popular option with our users.  2.7. Run Diagramtry to recreate this: https://stackoverflow.com/questions/43367690/how-to-get-gantt-plot-using-matplotlib<jupyter_code>## event_time and page

def get_user_run_diagram(user):
    cols = ['event_time', 'page', 'sessionId', 'itemInSession']
    user_run_diagram = df[df['userId'] == user][cols].copy()
    return user_run_diagram

def get_all_user_run_diagram():
    cols = ['event_time', 'page', 'sessionId', 'itemInSession']
    user_run_diagram = df[cols].copy()
    return user_run_diagram
def is_churn_user(user):
    if df[df.userId == user].churn_user.iloc[0] == 0:
        return "active"
    else:
        return "churn"<jupyter_output><empty_output><jupyter_text>2.7.1. Run Diagram for all users<jupyter_code>urd = get_all_user_run_diagram()
urd.index = pd.to_datetime(urd.event_time)

# count the hourly activity levels for the user
diag = urd.groupby(pd.Grouper(freq='H')).count()
# drop the redundant columns
diag = diag.drop(columns=['page', 'sessionId', 'itemInSession'])

diag['hour'] = diag.index.hour
diag['date'] = diag.index.date

fig, ax = plt.subplots(figsize=(18,10))
sns.heatmap(diag.pivot(index='hour', columns='date', values='event_time'))
ax.set_title("Run diagram of all users on Sparkify");<jupyter_output><empty_output><jupyter_text>From the above diagram, we can see that - overall - daytime hours are busier than nighttime ones, and Fridays & Saturdays (e.g. 5th and 6th Oct 2018) tend to be quieter, presumaby since our users will probably be going out and not near their desktops/laptops.We can also see that we seem to have a general decrease near the end of the month, followed by a surge - presumably at the start of a trial session, etc. - not possible to determine the reason with the current dataset.  2.7.2. Run diagram for a low-activity user<jupyter_code># user 78 or 79 (was 11)
user = '78'

urd = get_user_run_diagram(user)
has_user_churned = is_churn_user(user)
#urd = get_all_user_run_diagram()
urd.index = pd.to_datetime(urd.event_time)

# count the hourly activity levels for the user
diag = urd.groupby(pd.Grouper(freq='H')).count()
# drop the redundant columns
diag = diag.drop(columns=['page', 'sessionId', 'itemInSession'])

diag['hour'] = diag.index.hour
diag['date'] = diag.index.date

fig, ax = plt.subplots(figsize=(18,10))
sns.heatmap(diag.pivot(index='hour', columns='date', values='event_time'))
ax.set_title("user {} ({})".format(user, has_user_churned));<jupyter_output><empty_output><jupyter_text>From the above diagram we can see that a typical low-activity user just has a few short sessions of using Sparkify, without any particular pattern to the usage.  2.7.3. Run Diagram for a medium-activity user<jupyter_code># user 200014 or 155
user = '200014'

urd = get_user_run_diagram(user)
has_user_churned = is_churn_user(user)
#urd = get_all_user_run_diagram()
urd.index = pd.to_datetime(urd.event_time)

# count the hourly activity levels for the user
diag = urd.groupby(pd.Grouper(freq='H')).count()
# drop the redundant columns
diag = diag.drop(columns=['page', 'sessionId', 'itemInSession'])

diag['hour'] = diag.index.hour
diag['date'] = diag.index.date

fig, ax = plt.subplots(figsize=(18,10))
sns.heatmap(diag.pivot(index='hour', columns='date', values='event_time'))
ax.set_title("user {} ({})".format(user, has_user_churned));
user = '155'

urd = get_user_run_diagram(user)
has_user_churned = is_churn_user(user)
#urd = get_all_user_run_diagram()
urd.index = pd.to_datetime(urd.event_time)

# count the hourly activity levels for the user
diag = urd.groupby(pd.Grouper(freq='H')).count()
# drop the redundant columns
diag = diag.drop(columns=['page', 'sessionId', 'itemInSession'])

diag['hour'] = diag.index.hour
diag['date'] = diag.index.date

fig, ax = plt.subplots(figsize=(18,10))
sns.heatmap(diag.pivot(index='hour', columns='date', values='event_time'))
ax.set_title("user {} ({})".format(user, has_user_churned));<jupyter_output><empty_output><jupyter_text>For the above two diagrams we can see that a medium-activity user tends to have a few long sessions lasting a few hours, and they seem to appear around outside of typical 9-5 working hours.   2.7.4. Run Diagram for a high-activity user<jupyter_code># user 39 or 92
# Run the below line to get a list of users along with their activity levels - should
# you want to try the run diagram on any other users
#df.groupby('userId').status.count().sort_values(ascending = False)[150:250]

user = '92'

urd = get_user_run_diagram(user)
has_user_churned = is_churn_user(user)
#urd = get_all_user_run_diagram()
urd.index = pd.to_datetime(urd.event_time)

# count the hourly activity levels for the user
diag = urd.groupby(pd.Grouper(freq='H')).count()
# drop the redundant columns
diag = diag.drop(columns=['page', 'sessionId', 'itemInSession'])

diag['hour'] = diag.index.hour
diag['date'] = diag.index.date

fig, ax = plt.subplots(figsize=(18,10))
sns.heatmap(diag.pivot(index='hour', columns='date', values='event_time'))
ax.set_title("user {} ({})".format(user, has_user_churned));<jupyter_output><empty_output><jupyter_text>User 92 is quite keen on the Sparkify service and listens to music all the time! We have a lot of daytime activity and (hypothetically) a lot of nighttime usage just for background music whilst sleeping. We can infer from the above that length of sessions is something that matters when it comes to whether a user wants to keep using the service or not and would be useful to include this in one form or another in our modelling table. The machine/ browser that the user uses is going to be important as well - so we'll add them as features in addition.    3. 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.  Creating a summary table for the model - high level planFeatures:- userId- Device used (Mac, PC, Ipad, etc.)- browser used- number of sessions- gender- have they ever downgraded- have they added a friend- have they thumbs up or thumbs down a song- churnI considered a few extra features that I ultimately decided against - the code for them is below, for future reference.<jupyter_code>log_data.printSchema()
log_data.select('userId').distinct().rdd
ml_table = spark.createDataFrame(log_data.select('userId').distinct().rdd)
ml_table.collect()<jupyter_output><empty_output><jupyter_text>Feature 1: Days Active (Depreciated)This feature shows the number of days the user has been active.create date column in log_dataaggregate on userId by count<jupyter_code>convert_ts = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).strftime("%Y-%m-%d"))
log_data = log_data.withColumn('event_day', convert_ts('ts'))
import pyspark.sql.functions as func
days_logged_in = log_data.groupBy("userId").agg(func.countDistinct("event_day"))
days_logged_in.show()
ml_table = ml_table\
.join(days_logged_in, ml_table.userId == days_logged_in.userId)\
.select(ml_table["*"], days_logged_in["count(DISTINCT event_day)"])
ml_table.show()<jupyter_output>+------+-------------------------+
|userId|count(DISTINCT event_day)|
+------+-------------------------+
|100010|                        7|
|200002|                        7|
|   125|                        1|
|    51|                       13|
|   124|                       33|
|     7|                        8|
|    54|                       31|
|    15|                       19|
|   155|                        8|
|   132|                       21|
|   154|                        3|
|100014|                        6|
|   101|                       13|
|    11|                       16|
|   138|                       17|
|300017|                       50|
|    29|                       29|
|    69|                       12|
|100021|                        5|
|    42|                       44|
+------+-------------------------+
only showing top 20 rows<jupyter_text>Feature 2: Devices Usedjoin with the above parced dataframe (joining spark df with pandas df?)group by userId and drop duplicate devices and browsers<jupyter_code>log_data.select('userAgent').collect()
userAgents.head()
create_browser = udf(lambda x: userAgents[userAgents['index'] == x].browser.iloc[0])
create_machine = udf(lambda x: userAgents[userAgents['index'] == x].machine.iloc[0])
log_data = log_data.withColumn("browser", create_browser("userAgent"))
log_data = log_data.withColumn("machine", create_machine("userAgent"))
user_browsers = log_data\
.groupBy("userId", "browser")\
.agg(func.countDistinct("browser"))
ml_table = ml_table\
.join(user_browsers, ml_table.userId == user_browsers.userId)\
.select(ml_table["*"], user_browsers["browser"], user_browsers["count(DISTINCT browser)"])
user_machine = log_data\
.groupBy("userId", "machine")\
.agg(func.countDistinct("machine"))
ml_table = ml_table\
.join(user_machine, ml_table.userId == user_machine.userId)\
.select(ml_table["*"], user_machine["machine"], user_machine["count(DISTINCT machine)"])<jupyter_output><empty_output><jupyter_text>Feature 4: Number of sessions<jupyter_code>log_data.groupBy("userId").agg({"itemInSession": "count"}).show()

w = Window.partitionBy('userId')
log_data = log_data.select(log_data["*"], f.count('sessionId').over(w).alias("numberOfSessions"))
log_data = log_data.select(log_data["*"], f.count('itemInSession').over(w).alias("numberOfSessionItems"))<jupyter_output><empty_output><jupyter_text>Feature 5: Added a Friend<jupyter_code>users_added_friend = set(log_data\
.select("userId", "page").sort("userId")\
.dropDuplicates()\
.filter(log_data["page"] == "Add Friend")\
.select("userId")\
.toPandas()\
.userId)
# create a udf
added_friend = udf(lambda x: 1 if x in users_added_friend else 0, IntegerType())
log_data = log_data.withColumn('has_added_friend', added_friend('userId'))<jupyter_output><empty_output><jupyter_text>Feature 6: Thumbs Up/Down<jupyter_code>user_thumbs_up_down = set(log_data\
.select("userId", "page").sort("userId")\
.dropDuplicates()\
.filter((log_data["page"] == "Thumbs Down") | (log_data["page"] == "Thumbs Up") )\
.select("userId")\
.toPandas()\
.userId)
# create a udf
thumbs_up_down = udf(lambda x: 1 if x in user_thumbs_up_down else 0, IntegerType())
log_data = log_data.withColumn('thumbs_up_down', thumbs_up_down('userId'))
#log_data.take(2)
test = log_data.select("userId", "gender", "browser", "machine", "numberOfSessionItems",\
                       "has_added_friend", "thumbs_up_down", "downgrade_user", "churn_user").dropDuplicates()
#test.sort("userId").take(2)
test.toPandas().to_csv("ml_table.csv", index=False)<jupyter_output><empty_output><jupyter_text>ModellingSplit 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.<jupyter_code>ml_table = spark.read.option("header", "true").csv("ml_table.csv")
ml_table.printSchema()<jupyter_output>root
 |-- userId: string (nullable = true)
 |-- gender: string (nullable = true)
 |-- browser: string (nullable = true)
 |-- machine: string (nullable = true)
 |-- numberOfSessionItems: string (nullable = true)
 |-- has_added_friend: string (nullable = true)
 |-- thumbs_up_down: string (nullable = true)
 |-- downgrade_user: string (nullable = true)
 |-- churn_user: string (nullable = true)<jupyter_text>One-hot encode the categorical features<jupyter_code>#train_df, test_df = ml_table.randomSplit([0.8, 0.2], seed=42)
browser_vals = ml_table.select("browser").dropDuplicates().toPandas().browser
machine_vals = ml_table.select("machine").dropDuplicates().toPandas().machine
gender_vals = ml_table.select("gender").dropDuplicates().toPandas().gender
# browser - 14 vals
for browser_type in browser_vals.values: 
    browser_fn = udf(lambda x: 1 if x  == browser_type else 0, IntegerType())
    ml_table = ml_table.withColumn("browser_{}".format(browser_type), browser_fn("browser"))
    
# machine - 13 vals
for machine_type in machine_vals.values: 
    machine_fn = udf(lambda x: 1 if x  == machine_type else 0, IntegerType())
    ml_table = ml_table.withColumn("machine_{}".format(machine_type), machine_fn("machine"))
    
# gender
for gender_type in gender_vals.values: 
    gender_fn = udf(lambda x: 1 if x  == gender_type else 0, IntegerType())
    ml_table = ml_table.withColumn("gender_{}".format(gender_type), gender_fn("gender"))
ml_table.take(2)
ml_table.printSchema()
# remove the gender col
ml_table = ml_table.drop(ml_table["gender"])

# remove the browser col
ml_table = ml_table.drop(ml_table["browser"])

# remove the machine col
ml_table = ml_table.drop(ml_table["machine"])
ml_table.take(1)
ml_table = ml_table.withColumn("numberOfSessionItems", ml_table["numberOfSessionItems"].cast(IntegerType()))

ml_table = ml_table.withColumn("has_added_friend", ml_table["has_added_friend"].cast(IntegerType()))

ml_table = ml_table.withColumn("thumbs_up_down", ml_table["thumbs_up_down"].cast(IntegerType()))

ml_table = ml_table.withColumn("downgrade_user", ml_table["downgrade_user"].cast(IntegerType()))

ml_table = ml_table.withColumn("churn_user", ml_table["churn_user"].cast(IntegerType()))
ml_table = ml_table.withColumnRenamed("churn_user", "label")
train_df, test_df = ml_table.randomSplit([0.8, 0.2], seed=42)
ml_table.columns

input_columns = [
 'numberOfSessionItems',
 'has_added_friend',
 'thumbs_up_down',
 'downgrade_user',
 #'churn_user', -> this is our target var
 'browser_Chrome 35',
 'browser_IE 10',
 'browser_Firefox 30',
 'browser_IE 9',
 'browser_Chrome 37',
 'browser_Firefox 32',
 'browser_Firefox 31',
 'browser_Chrofme 36',
 'browser_Chrome 36',
 'browser_IE 11',
 'browser_Safari 6_1',
 'browser_Firefox 24',
 'browser_Safari 7',
 'browser_Safari 8',
 'machine_Windows 8',
 'machine_iPad',
 'machine_WIndows Vista',
 'machine_Mountain Lion',
 'machine_Windows 8_1',
 'machine_iPhone',
 'machine_Linux',
 'machine_Mavericks',
 'machine_Yosemite',
 'machine_Lion',
 'machine_Windows XP',
 'machine_Windows 7',
 'machine_Snow Leopard',
 'gender_F',
 'gender_M'
]
accuracy_scores = []
f1_scores = []
assembler = VectorAssembler(inputCols=input_columns, outputCol="features")
scalar = MinMaxScaler(inputCol="features", outputCol="scaled_features")

### Logistic Regression

log_reg =  LogisticRegression(featuresCol="scaled_features", maxIter=10, regParam=0.0, elasticNetParam=0)
pipeline_log_reg = Pipeline(stages=[assembler, scalar, log_reg])
paramGrid = ParamGridBuilder().addGrid(log_reg.regParam,[0.0, 0.1]).build()
crossval_lr = CrossValidator(estimator=pipeline_log_reg, estimatorParamMaps=paramGrid,\
                             evaluator=MulticlassClassificationEvaluator(),
                             numFolds=3)
model_log_reg = crossval_lr.fit(train_df)
results_log_reg = model_log_reg.transform(test_df)
# Append the log regression acc score
accuracy_scores.append(results_log_reg.filter(results_log_reg.label == results_log_reg.prediction).count() / results_log_reg.count())
f1_evaluator = MulticlassClassificationEvaluator(metricName="f1")

# Append the log regression f1 score
f1_scores.append(f1_evaluator.evaluate(results_log_reg))

### Random Forrest Classifier

rfc = RandomForestClassifier(featuresCol="scaled_features", numTrees=2)
pipeline_rfc = Pipeline(stages=[assembler, scalar, rfc])
paramGrid = ParamGridBuilder().addGrid(rfc.numTrees,[2,4,6,8,10]).build()

crossval_rf = CrossValidator(estimator=pipeline_rfc,
                          estimatorParamMaps=paramGrid,
                          evaluator=MulticlassClassificationEvaluator(),
                          numFolds=3)

model_rfc = crossval_rf.fit(train_df)
results_rfc = model_rfc.transform(test_df)
accuracy_scores.append(results_rfc.filter(results_rfc.label == results_rfc.prediction).count() / results_rfc.count())
f1_scores.append(f1_evaluator.evaluate(results_rfc))

## GBT

gbt = GBTClassifier(labelCol="label", featuresCol="scaled_features", maxIter=10)
pipeline_gbt = Pipeline(stages=[assembler, scalar, gbt])
paramGrid = ParamGridBuilder().addGrid(gbt.maxIter,[5, 10, 15]).build()
crossval_gbt = CrossValidator(estimator=pipeline_gbt,
                          estimatorParamMaps=paramGrid,
                          evaluator=MulticlassClassificationEvaluator(),
                          numFolds=3, collectSubModels=True)
model_gbt = crossval_gbt.fit(train_df)
results_gbt = model_gbt.transform(test_df)
accuracy_scores.append(results_gbt.filter(results_gbt.label == results_gbt.prediction).count()/ results_gbt.count())
f1_scores.append(f1_evaluator.evaluate(results_gbt))<jupyter_output><empty_output><jupyter_text>Results and Validation Note: In terms of refinement, we varied one parameter for each model and we will use the CrossValidator in Spark with the MulticlassClassificationEvaluator to determine for each model which parameter configuration works best. - In the Logistic Regression, we will vary the regParam between 0.0 and 0.1.- For the Random Forest Classifier, we will vary the numTrees parameter for 2,4,6,8 and 10. - For the GBT Classifier, we will vary the maxIter and try 5, 10 and 15.Below is a printout of what configuration worked best for each model.<jupyter_code>model_log_reg.explainParams()
model_rfc.explainParams()
model_gbt.explainParams()<jupyter_output><empty_output><jupyter_text>F1 and accuracy scores<jupyter_code>accuracy_scores
f1_scores
sns.barplot(y=accuracy_scores, x = ['Logistic Reg', 'Random Forest', 'GBT'])

plt.gca().set_title("Accuracy Scores for the trained models");
sns.barplot(y=f1_scores, x = ['Logistic Reg', 'Random Forest', 'GBT'])

plt.gca().set_title("F1-scores for the trained models");
gbt_feature_scores = [model_gbt.bestModel.stages[2].featureImportances[i] for i in range(0,33)]
#depreciated
#rfc_feature_scores = [model_rfc.bestModel.stages[2].featureImportances[i] for i in range(0,33)]
feature_importance = pd.DataFrame({'feature': input_columns, 'score': gbt_feature_scores}).sort_values(by="score", ascending=False)
feature_importance[0:10]
sns.barplot(data=feature_importance, y='feature', x='score')
f = plt.gcf()
f.set_size_inches(16,12)
plt.gca().set_title("Feature Importance of the GBT Model");<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13459608307383483, 'fraction_numerical': 0.05602594127184291, 'mean_word_length': 4.50318651749044, 'pattern_counts': {'":': 1, '<': 109, '<?xml version=': 0, '>': 109, 'https://': 4, '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': '28', 'dir': 'jupyter-structured-clean-dedup', 'id': '633149', 'license': 'MIT', 'n_tokens_mistral': 14750, 'n_tokens_neox': 13246, 'n_words': 3647, 'path': 'Sparkify.ipynb', 'repo_name': 'martinkirov/dsnd-capstone', 'size': '38857'}