<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
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
%matplotlib inline

from pyspark.sql import SparkSession
from pyspark.sql.window import Window
from pyspark.sql.functions import asc, col, concat, count, desc, isnull, lag, row_number, udf, sumDistinct
from pyspark.sql.functions import sum as Fsum, max as Fmax
from pyspark.sql.types import IntegerType, StringType
import datetime

from pyspark.ml.classification import LogisticRegression, RandomForestClassifier
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import VectorAssembler, StandardScaler, StringIndexer
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from sklearn.metrics import f1_score, precision_score, roc_auc_score, recall_score
from pyspark.ml import Pipeline

import re
# create a Spark session
spark = SparkSession \
    .builder \
    .appName("Sparkify") \
    .getOrCreate()<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>user_log = spark.read.json("mini_sparkify_event_data.json")
user_log.printSchema()
# len of user_log
user_log.count()
# check missing values in sessionId
user_log.filter(isnull(user_log['sessionId'])).count()
# check missing values in userId
user_log.filter(isnull(user_log['userId'])).count()
# user_log to Pandas
user_log_pd = user_log.toPandas()
user_log_pd.info()
# how many of each page event are there
user_log_pd.page.value_counts()<jupyter_output><empty_output><jupyter_text>NextSong is in the top of the page events. Cancellation Confirmation just have 52.<jupyter_code># statistics of `sessionId`
user_log.describe("sessionId").show()
# statistics of `userId`
user_log.describe("userId").show()
# some info about a userId = 99 (Madison)
user_log.select(['userId', 'firstName', 'page', 'song']).where(user_log.userId == '99').collect()
# drop duplicates in `page`
user_log.select("page").dropDuplicates().sort("page").show()
# timestamp to hour
get_hour = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0). hour)
user_log = user_log.withColumn("hour", get_hour(user_log.ts))
user_log.head()
# timestamp to day
get_day = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0). day)
user_log = user_log.withColumn("day", get_day(user_log.ts))
user_log.head()
# songs in hour
songs_in_hour = user_log.filter(user_log.page == "NextSong").groupby(user_log.hour).count().orderBy(user_log.hour.cast("float"))
songs_in_hour.show()
# transform to Pandas dataframe
songs_in_hour_pd = songs_in_hour.toPandas()
songs_in_hour_pd.hour = pd.to_numeric(songs_in_hour_pd.hour)
# plot songs played by hour
plt.scatter(songs_in_hour_pd['hour'], songs_in_hour_pd['count'])
plt.title("Songs played by Hour")
plt.xlabel("Hour")
plt.ylabel("Songs played");
plt.grid();
# songs in days
songs_in_days = user_log.filter(user_log.page == "NextSong").groupby(user_log.day).count().orderBy(user_log.day.cast("float"))
songs_in_days.show()
# transform to Pandas dataframe
songs_in_days_pd = songs_in_days.toPandas()
songs_in_days_pd.day = pd.to_numeric(songs_in_days_pd.day)
# plot songs played by day
plt.scatter(songs_in_days_pd['day'], songs_in_days_pd['count'])
plt.title("Songs played by Day")
plt.xlabel("Day")
plt.ylabel("Songs played");
plt.grid();
# drop rows with missing values
user_log_valid = user_log.dropna(how = "any", subset = ["userId", "sessionId"])
user_log_valid.count()
user_log_valid = user_log_valid.filter(user_log_valid["userId"] != "")
user_log_valid.count()<jupyter_output><empty_output><jupyter_text>Only empty strings as `userId` were removed. There wasn't missing values in `userId` or `sessionId` columns.   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># filter by `Cancellation Confirmation`
user_log_valid.filter("page == 'Cancellation Confirmation'").show()
# number of cancellations
user_log_valid.filter("page == 'Cancellation Confirmation'").select('userId').dropDuplicates().count()
# Kayla user service activity
Kayla_log = user_log_valid.select(['userId', 'sessionId', 'firstName', 'page', 'level', 'song', 'hour']).where(user_log.userId == '17')
Kayla_log.show()
# churn event
flag_churn_event = udf(lambda x: 1 if x == "Cancellation Confirmation" else 0, IntegerType())
user_churn = user_log_valid.withColumn("pre_churn", flag_churn_event("page"))
user_churn = user_churn.withColumn('churn', Fmax('pre_churn').over(Window.partitionBy('userId')))
user_churn = user_churn.dropDuplicates(['userId'])
# table of churn
user_churn.groupBy('churn').count().show()
# user_churn to Pandas
user_churn_pd = user_churn.toPandas()
# hist of churn
user_churn_pd['churn'].value_counts().plot(kind="bar", color=["lightblue", "salmon"])
plt.title("Histogram of Churn")
plt.xlabel("Churn")
plt.ylabel("Count")
plt.grid();<jupyter_output><empty_output><jupyter_text>There are 52 churns and 173 no-churn users.<jupyter_code># table of gender
user_churn.groupBy('gender').count().show()
# hist of gender
user_churn_pd['gender'].value_counts().plot(kind="bar", color=["lightblue", "salmon"])
plt.title("Histogram of Gender")
plt.xlabel("Gender")
plt.ylabel("Count")
plt.grid();<jupyter_output><empty_output><jupyter_text>There are 104 female and 121 male.<jupyter_code># table of level
user_churn_pd['level'].value_counts()
# plot of level
user_churn_pd['level'].value_counts().plot(kind="bar", color=["lightblue", "salmon"])
plt.title("Histogram of Level")
plt.xlabel("Level")
plt.ylabel("Count")
plt.grid();<jupyter_output><empty_output><jupyter_text>There are 177 free and 48 paid.<jupyter_code># table of level vs gender
pd.crosstab(user_churn_pd.level, user_churn_pd.gender)
# plot the hist of level vs gender
pd.crosstab(user_churn_pd.level, user_churn_pd.gender).plot(kind="bar",
                                    color=["salmon", "lightblue"])
plt.title("Histogram of the Level vs Gender")
plt.xlabel("Level")
plt.ylabel("Count")
plt.grid();

# table of churn vs gender
pd.crosstab(user_churn_pd.churn, user_churn_pd.gender)
# hist of churn vs gender
pd.crosstab(user_churn_pd.churn, user_churn_pd.gender).plot(kind="bar",
                                    color=["salmon", "lightblue"])
plt.title("Histogram of the Churn vs Gender")
plt.xlabel("Churn")
plt.ylabel("Count")
plt.grid();
# table of churn vs level
pd.crosstab(user_churn_pd.churn, user_churn_pd.level)
# hist of churn vs gender
pd.crosstab(user_churn_pd.churn, user_churn_pd.level).plot(kind="bar",
                                    color=["salmon", "lightblue"])
plt.title("Histogram of the Churn vs Level")
plt.xlabel("Churn")
plt.ylabel("Count")
plt.grid();
# table of location - topten
user_churn_pd['location'].value_counts().head(10)
# hist of location
user_churn_pd['location'].value_counts().head(10).plot(kind="bar")
plt.title("Histogram of TopTen Locations")
plt.xlabel("Location")
plt.ylabel("Count")
plt.grid();<jupyter_output><empty_output><jupyter_text>Los Angeles, New York and Phoenix are in the top three cities.<jupyter_code># create temp table
user_log_valid.createOrReplaceTempView("log_table")

# create churn column
churn_df = spark.sql(
'''
    SELECT users.userId, 
    CASE WHEN CAST(churned.userId AS INT) IS NULL THEN 0
    ELSE 1 END AS churned
    FROM (
        SELECT DISTINCT CAST(userId AS INT)
        FROM log_table
        ORDER BY userId
    ) AS users
    FULL OUTER JOIN (
        SELECT CAST(log_table.userId AS INT)
        FROM log_table
        WHERE page = 'Cancellation Confirmation'
        ORDER BY userId
    ) AS churned
    ON users.userId = churned.userId
    ORDER BY userId
'''
)

churn_df.show()
# how many churn (1) / not-churn(0) are there
churn_df.groupBy('churned').count().show()
# churn_df size
churn_df.count()<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># pages available

spark.sql('''
          SELECT DISTINCT page
          FROM log_table 
          ORDER BY page ASC
          '''
          ).show()<jupyter_output>+--------------------+
|                page|
+--------------------+
|               About|
|          Add Friend|
|     Add to Playlist|
|              Cancel|
|Cancellation Conf...|
|           Downgrade|
|               Error|
|                Help|
|                Home|
|              Logout|
|            NextSong|
|         Roll Advert|
|       Save Settings|
|            Settings|
|    Submit Downgrade|
|      Submit Upgrade|
|         Thumbs Down|
|           Thumbs Up|
|             Upgrade|
+--------------------+<jupyter_text>Feature 1:<jupyter_code># table to create a boolean feature related to gender
gender = spark.sql(
    '''
    SELECT CAST(userId AS INT),
    CASE WHEN gender = 'M' THEN 0 ELSE 1 END AS gender
    FROM log_table
    GROUP BY userId, gender
    ORDER BY userId, gender
    '''
)

gender.show(10)
# join gender with churn_df
churn_df = churn_df.join(gender, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+
|userId|churned|gender|
+------+-------+------+
|     2|      0|     1|
|     3|      1|     0|
|     4|      0|     0|
|     5|      0|     0|
|     6|      0|     0|
|     7|      0|     0|
|     8|      0|     1|
|     9|      0|     0|
|    10|      0|     0|
|    11|      0|     1|
|    12|      1|     1|
|    13|      0|     1|
|    14|      0|     0|
|    15|      0|     0|
|    16|      0|     1|
|    17|      1|     1|
|    18|      1|     0|
|    19|      0|     1|
|    20|      0|     1|
|    21|      0|     0|
+------+-------+------+
only showing top 20 rows<jupyter_text>Feature 2:<jupyter_code># table to count the number of total add friends per userId
total_friends = spark.sql(
    '''
    WITH h AS (
        SELECT CAST(userId AS INT)
        FROM log_table
        WHERE page = 'Add Friend'
        ORDER BY userId
    )
    SELECT 
        h.userId,
        COUNT(*) AS total_friends
    FROM h
    GROUP BY userId
    ORDER BY userId
    '''
)

total_friends.show(10)
# join total_friends with churn_df
churn_df = churn_df.join(total_friends, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+
|userId|churned|gender|total_friends|
+------+-------+------+-------------+
|     2|      0|     1|           20|
|     3|      1|     0|            1|
|     4|      0|     0|           46|
|     5|      0|     0|            3|
|     6|      0|     0|           41|
|     7|      0|     0|            1|
|     8|      0|     1|            5|
|     9|      0|     0|           40|
|    10|      0|     0|           12|
|    11|      0|     1|            6|
|    12|      1|     1|           13|
|    13|      0|     1|           32|
|    14|      0|     0|           19|
|    15|      0|     0|           31|
|    16|      0|     1|           10|
|    17|      1|     1|           12|
|    18|      1|     0|           10|
|    19|      0|     1|            4|
|    20|      0|     1|           25|
|    21|      0|     0|            8|
+------+-------+------+-------------+
only showing top 20 rows<jupyter_text>Feature 3:<jupyter_code># table to count the number of total hours per userId
hours_user = spark.sql(
'''
    WITH h AS (
        SELECT
            CAST(userId AS INT),
            CAST(hour AS INT) AS hour
        FROM log_table
        ORDER BY userId, hour
    )
    SELECT 
        h.userId,
        COUNT(h.hour) AS total_hours
    FROM h
    GROUP BY userId
    ORDER BY userId
'''
)

hours_user.show()
# join hours_user with churn_df
churn_df = churn_df.join(hours_user, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+
|userId|churned|gender|total_friends|total_hours|
+------+-------+------+-------------+-----------+
|     2|      0|     1|           20|        899|
|     3|      1|     0|            1|        254|
|     4|      0|     0|           46|       2442|
|     5|      0|     0|            3|        218|
|     6|      0|     0|           41|       3761|
|     7|      0|     0|            1|        201|
|     8|      0|     1|            5|        334|
|     9|      0|     0|           40|       3191|
|    10|      0|     0|           12|        795|
|    11|      0|     1|            6|        848|
|    12|      1|     1|           13|       1064|
|    13|      0|     1|           32|       1654|
|    14|      0|     0|           19|       1432|
|    15|      0|     0|           31|       2278|
|    16|      0|     1|           10|        819|
|    17|      1|     1|           12|       1102|
|    18|      1|     0|           10|        513|
[...]<jupyter_text>Feature 4:<jupyter_code># table to count the number of total days per userId
days_user = spark.sql(
'''
    WITH d AS (
        SELECT DISTINCT
            CAST(userId AS INT),
            DATE(TIMESTAMP(ts)) AS date
        FROM log_table
        ORDER BY userId, date
    )
    SELECT 
        d.userId,
        COUNT(d.date) AS total_days
    FROM d
    GROUP BY userId
    ORDER BY userId
'''
)

days_user.show()
# join days_user with churn_df
churn_df = churn_df.join(days_user, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+
|userId|churned|gender|total_friends|total_hours|total_days|
+------+-------+------+-------------+-----------+----------+
|     2|      0|     1|           20|        899|       782|
|     3|      1|     0|            1|        254|       230|
|     4|      0|     0|           46|       2442|      2133|
|     5|      0|     0|            3|        218|       171|
|     6|      0|     0|           41|       3761|      3335|
|     7|      0|     0|            1|        201|       170|
|     8|      0|     1|            5|        334|       259|
|     9|      0|     0|           40|       3191|      2760|
|    10|      0|     0|           12|        795|       699|
|    11|      0|     1|            6|        848|       667|
|    12|      1|     1|           13|       1064|       955|
|    13|      0|     1|           32|       1654|      1364|
|    14|      0|     0|           19|       1432|      1244|
|    15|      0|     0| [...]<jupyter_text>Feature 5:<jupyter_code># table to count the number of total songs per userId
songs_user = spark.sql(
    '''
    WITH s AS (
        SELECT CAST(userId AS INT)
        FROM log_table
        WHERE page = 'NextSong'
        ORDER BY userId
    )
    SELECT 
        s.userId,
        COUNT(*) AS total_songs
    FROM s
    GROUP BY userId
    ORDER BY userId
    '''
)

songs_user.show()
# join songs_user with churn_df
churn_df = churn_df.join(songs_user, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+-----------+
|userId|churned|gender|total_friends|total_hours|total_days|total_songs|
+------+-------+------+-------------+-----------+----------+-----------+
|     2|      0|     1|           20|        899|       782|        755|
|     3|      1|     0|            1|        254|       230|        214|
|     4|      0|     0|           46|       2442|      2133|       2048|
|     5|      0|     0|            3|        218|       171|        161|
|     6|      0|     0|           41|       3761|      3335|       3159|
|     7|      0|     0|            1|        201|       170|        150|
|     8|      0|     1|            5|        334|       259|        251|
|     9|      0|     0|           40|       3191|      2760|       2676|
|    10|      0|     0|           12|        795|       699|        673|
|    11|      0|     1|            6|        848|       667|        647|
|    12|      1|     1|           13|       1064|  [...]<jupyter_text>Feature 6:<jupyter_code># table to count the ratio of interactions per hour per day
interact_ratio_avg = spark.sql(
    '''
    WITH i AS (
        SELECT
            CAST(userId AS INT),
            DATE_TRUNC('HOUR', DATE(TIMESTAMP(ts))) AS hour,
            COUNT(*) AS num_interact
        FROM log_table
        GROUP BY userId, 2
        ORDER BY userId, 2
    )
    SELECT 
        i.userId,
        ROUND(AVG(i.num_interact), 2) AS interact_ratio
    FROM i
    GROUP BY userId
    ORDER BY userId
    '''
)

interact_ratio_avg.show()
# join interact_ratio_avg with churn_df
churn_df = churn_df.join(interact_ratio_avg, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+-----------+--------------+
|userId|churned|gender|total_friends|total_hours|total_days|total_songs|interact_ratio|
+------+-------+------+-------------+-----------+----------+-----------+--------------+
|     2|      0|     1|           20|        899|       782|        755|          1.15|
|     3|      1|     0|            1|        254|       230|        214|           1.1|
|     4|      0|     0|           46|       2442|      2133|       2048|          1.14|
|     5|      0|     0|            3|        218|       171|        161|          1.27|
|     6|      0|     0|           41|       3761|      3335|       3159|          1.13|
|     7|      0|     0|            1|        201|       170|        150|          1.18|
|     8|      0|     1|            5|        334|       259|        251|          1.29|
|     9|      0|     0|           40|       3191|      2760|       2676|          1.16|
|    10|      0|     0|         [...]<jupyter_text>Feature 7:<jupyter_code># table to count the number of days since join
join_days = spark.sql(
    '''
    WITH t2 AS (
        SELECT 
            t1.userId,
            t1.date,
            ROW_NUMBER() OVER (
                PARTITION BY t1.userId
                ORDER BY t1.date ASC
            ) AS row_num_1,
            ROW_NUMBER() OVER (
                PARTITION BY t1.userId
                ORDER BY t1.date DESC
            ) AS row_num_2
        FROM (
            SELECT DISTINCT
                CAST(log_table.userId AS INT) AS userId,
                DATE(TIMESTAMP(log_table.ts)) AS date
            FROM log_table
            ORDER BY 1, 2
        ) AS t1
        ORDER BY 1, 2
    ),
    t3 AS (
        SELECT
            t2.userId,
            t2.date
        FROM t2
        WHERE t2.row_num_1 = 1
    ),
    t4 AS (
        SELECT
            t2.userId,
            t2.date
        FROM t2
        WHERE t2.row_num_2 = 1
    )
    SELECT
        t3.userId,
        DATEDIFF(t4.date, t3.date) AS days_since_joining
    FROM t3
    INNER JOIN t4
    ON t3.userId = t4.userId
    ORDER BY 1
    '''
)

join_days.show()
# join join_days with churn_df
churn_df = churn_df.join(join_days, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+
|userId|churned|gender|total_friends|total_hours|total_days|total_songs|interact_ratio|days_since_joining|
+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+
|   148|      0|     0|            7|        518|       422|        398|          1.23|             55372|
|    85|      0|     0|           58|       4370|      3833|       3616|          1.14|             59016|
|   137|      0|     0|            8|        209|       166|        154|          1.26|             54318|
|    65|      0|     0|           53|       2544|      2225|       2113|          1.14|             58729|
|200001|      1|     0|            0|        158|       124|        115|          1.27|             14189|
|    53|      1|     0|           25|       2064|      1778|       1746|          1.16|             48752|
|   133|      0|     0|            1|[...]<jupyter_text>Feature 8:<jupyter_code># table to count the number of thumbs down per day
thumbs_down = spark.sql(
    '''
    SELECT 
        CAST(userId AS INT) AS userId, 
        COUNT(*) AS thumbs_down
    FROM log_table
    WHERE page = 'Thumbs Down'
    GROUP BY CAST(userId AS INT)
    ORDER BY CAST(userId AS INT)
    '''
)

thumbs_down.show()
# join thumbs_down with churn_df
churn_df = churn_df.join(thumbs_down, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+
|userId|churned|gender|total_friends|total_hours|total_days|total_songs|interact_ratio|days_since_joining|thumbs_down|
+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+
|   148|      0|     0|            7|        518|       422|        398|          1.23|             55372|          3|
|    85|      0|     0|           58|       4370|      3833|       3616|          1.14|             59016|         38|
|   137|      0|     0|            8|        209|       166|        154|          1.26|             54318|          1|
|    65|      0|     0|           53|       2544|      2225|       2113|          1.14|             58729|         17|
|200001|      1|     0|            0|        158|       124|        115|          1.27|             14189|          1|
|    53|      1|     0|           25|       2064[...]<jupyter_text>Feature 9:<jupyter_code># table to count the number of thumbs up per day
thumbs_up = spark.sql(
    '''
    SELECT 
        CAST(userId AS INT) AS userId, 
        COUNT(*) AS thumbs_up
    FROM log_table
    WHERE page = 'Thumbs Up'
    GROUP BY CAST(userId AS INT)
    ORDER BY CAST(userId AS INT)
    '''
)

thumbs_up.show()
# join thumbs_up with churn_df
churn_df = churn_df.join(thumbs_up, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+---------+
|userId|churned|gender|total_friends|total_hours|total_days|total_songs|interact_ratio|days_since_joining|thumbs_down|thumbs_up|
+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+---------+
|   148|      0|     0|            7|        518|       422|        398|          1.23|             55372|          3|       28|
|    85|      0|     0|           58|       4370|      3833|       3616|          1.14|             59016|         38|      192|
|   137|      0|     0|            8|        209|       166|        154|          1.26|             54318|          1|        8|
|    65|      0|     0|           53|       2544|      2225|       2113|          1.14|             58729|         17|      111|
|200001|      1|     0|            0|        158|       124|        115|          1.27|          [...]<jupyter_text>Feature 10:<jupyter_code># table to count the number of add to playlist per day
add_playlist = spark.sql(
    '''
    SELECT 
        CAST(userId AS INT) AS userId, 
        COUNT(*) AS add_playlist
    FROM log_table
    WHERE page = 'Add to Playlist'
    GROUP BY CAST(userId AS INT)
    ORDER BY CAST(userId AS INT)
    '''
)

add_playlist.show()
# join add_playlist with churn_df
churn_df = churn_df.join(add_playlist, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+---------+------------+
|userId|churned|gender|total_friends|total_hours|total_days|total_songs|interact_ratio|days_since_joining|thumbs_down|thumbs_up|add_playlist|
+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+---------+------------+
|   148|      0|     0|            7|        518|       422|        398|          1.23|             55372|          3|       28|           5|
|    85|      0|     0|           58|       4370|      3833|       3616|          1.14|             59016|         38|      192|         108|
|   137|      0|     0|            8|        209|       166|        154|          1.26|             54318|          1|        8|           4|
|    65|      0|     0|           53|       2544|      2225|       2113|          1.14|             58729|         17|      111|          68|
|20000[...]<jupyter_text>Feature 11:<jupyter_code># table to create a boolean feature related to the level
level = spark.sql(
    '''
    SELECT CAST(userId AS INT),
    CASE WHEN level = 'free' THEN 0 ELSE 1 END AS level
    FROM log_table
    GROUP BY userId, level
    ORDER BY userId, level
    '''
)

level.show(10)
# join level with churn_df
churn_df = churn_df.join(level, on='userId', how='left').fillna(0.0)
churn_df.show()<jupyter_output>+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+---------+------------+-----+
|userId|churned|gender|total_friends|total_hours|total_days|total_songs|interact_ratio|days_since_joining|thumbs_down|thumbs_up|add_playlist|level|
+------+-------+------+-------------+-----------+----------+-----------+--------------+------------------+-----------+---------+------------+-----+
|   148|      0|     0|            7|        518|       422|        398|          1.23|             55372|          3|       28|           5|    0|
|    85|      0|     0|           58|       4370|      3833|       3616|          1.14|             59016|         38|      192|         108|    1|
|    85|      0|     0|           58|       4370|      3833|       3616|          1.14|             59016|         38|      192|         108|    0|
|   137|      0|     0|            8|        209|       166|        154|          1.26|             54318|      [...]<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.<jupyter_code># features type
churn_df.dtypes
# VectorAssembler to combine features
assembler = VectorAssembler(
    inputCols=[col for col in churn_df.columns if col not in ['userId', 'churned']], 
    outputCol='unscaled_features')

data = assembler \
    .transform(churn_df) \
    .select(
        col('userId'),
        col('churned').alias('label'), 
        col('unscaled_features')
    )

# split the dataset
train, test = data.randomSplit([0.8, 0.2], seed=42)

# scaler
train_scaler = StandardScaler(
    inputCol='unscaled_features', 
    outputCol='features'
)

train = train_scaler.fit(train).transform(train)
test = train_scaler.fit(test).transform(test)<jupyter_output><empty_output><jupyter_text>Logistic Regression<jupyter_code># Init Logistic Regression
lr_clf = LogisticRegression()

lr_paramGrid = ParamGridBuilder() \
    .addGrid(lr_clf.regParam, [0.1, 0.01]) \
    .build()

# cross_validation
lr_crossval = CrossValidator(
    estimator=lr_clf,
    estimatorParamMaps=lr_paramGrid,
    evaluator=BinaryClassificationEvaluator(),
    numFolds=5
)

# fitting
lr_cvModel = lr_crossval.fit(train)

# predictions
lr_prediction = lr_cvModel.transform(test)
# save the LR model
lr_cvModel.save('./logistic_regression/')
# dataset to a pandas df
lr_prediction_df = lr_prediction.select(['userId', 'label', 'probability', 'prediction']).toPandas()

lr_user_churn_probabilities = lr_prediction_df[['userId', 'probability']]
lr_user_churn_probabilities.head()

# F1, precision, recall and AUC
lr_y_test_pred = lr_prediction_df['prediction']
lr_y_test_true = lr_prediction_df['label']

print({
    'f1_score': f1_score(y_true=lr_y_test_true, y_pred=lr_y_test_pred),
    'precision': precision_score(y_true=lr_y_test_true, y_pred=lr_y_test_pred),
    'recall': recall_score(y_true=lr_y_test_true, y_pred=lr_y_test_pred),
    'AUC': roc_auc_score(y_true=lr_y_test_true, y_score=lr_y_test_pred)
})<jupyter_output>{'f1_score': 0.42857142857142855, 'precision': 0.375, 'recall': 0.5, 'AUC': 0.66379310344827591}<jupyter_text>Random Forest<jupyter_code># Init Random Forest
rf_clf = RandomForestClassifier()

rf_paramGrid = ParamGridBuilder() \
    .addGrid(rf_clf.maxDepth, [2, 4, 6]) \
    .addGrid(rf_clf.maxBins, [20, 60]) \
    .addGrid(rf_clf.numTrees, [5, 20, 100]) \
    .build()

# cross-validation
rf_crossval = CrossValidator(
    estimator=rf_clf,
    estimatorParamMaps=rf_paramGrid,
    evaluator=BinaryClassificationEvaluator(),
    numFolds=5
)

# fitting
rf_cvModel = rf_crossval.fit(train)
# predictions
rf_prediction = rf_cvModel.transform(test)
# dataset to a pandas df
rf_prediction_df = rf_prediction.select(['userId', 'label', 'probability', 'prediction']).toPandas()

rf_user_churn_probabilities = rf_prediction_df[['userId', 'probability']]
rf_user_churn_probabilities.head()

# F1, precision, recall and AUC
rf_y_test_pred = rf_prediction_df['prediction']
rf_y_test_true = rf_prediction_df['label']

print({
    'f1_score': f1_score(y_true=rf_y_test_true, y_pred=rf_y_test_pred),
    'precision': precision_score(y_true=rf_y_test_true, y_pred=rf_y_test_pred),
    'recall': recall_score(y_true=rf_y_test_true, y_pred=rf_y_test_pred),
     'AUC': roc_auc_score(y_true=rf_y_test_true, y_score=rf_y_test_pred)
})
# best params.
best_param = rf_cvModel.bestModel

type(train)
train_df = train.toPandas()
type(train_df)
train_df.columns
# k-fold cross validation

# Import StratifiedKFold
from sklearn.model_selection import StratifiedKFold

# Create a StratifiedKFold object
str_kf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

# Loop through each split
fold = 0
for train_index, test_index in str_kf.split(train_df, train_df['label']):
    # Obtain training and testing folds
    cv_train, cv_test = train_df.iloc[train_index], train_df.iloc[test_index]
    print('Fold: {}'.format(fold))
    print('CV train shape: {}'.format(cv_train.shape))
    print('Churn in CV train: {}\n'.format(sum(cv_train.label == 0)))
    fold += 1<jupyter_output>Fold: 0
CV train shape: (232, 4)
Churn in CV train: 176

Fold: 1
CV train shape: (232, 4)
Churn in CV train: 176

Fold: 2
CV train shape: (232, 4)
Churn in CV train: 176

Fold: 3
CV train shape: (232, 4)
Churn in CV train: 176

Fold: 4
CV train shape: (232, 4)
Churn in CV train: 176
{'fraction_non_alphanumeric': 0.1659402050463503, 'fraction_numerical': 0.05282706120694947, 'mean_word_length': 1.830249326730953, 'pattern_counts': {'":': 0, '<': 79, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 74, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '48458', 'license': 'FTL', 'n_tokens_mistral': 11648, 'n_tokens_neox': 10626, 'n_words': 2723, 'path': 'Sparkify.ipynb', 'repo_name': 'kmera/Sparkify_Project', 'size': '32578'}