<jupyter_start><jupyter_text>Sparkify Capstone Project  Business UnderstandingChurn modeling is one of the common analytics problems faced by data scientists working with business teams. It impacts the revenue of the company directly. Churn modeling has some interesting constraints and implications for data scientists as the dataset is imbalanced and also model interpretability is an important requirement.In this project, we will look at user interaction logs for a fictional music streaming company Sparkify. The full dataset is 12GB and would need a spark cluster to work with. We have explored the mini(128 MB) locally and medium(230 MB) datasets on IBM Watson cluster. This project familiarizes us with PySpark library for handling Big Data. We walk through the entire data lifecycle in this project and also present the findings for technical audience in a Medium post.<jupyter_code># import libraries
from pyspark.sql import SparkSession
from pyspark.sql.functions import isnan, count, when, col, desc, udf, col, sort_array, asc, avg, lit
from pyspark.sql.functions import max as Fmax
from pyspark.sql.window import Window
from pyspark.sql.types import IntegerType

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression, GBTClassifier,LinearSVC, RandomForestClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import StandardScaler, VectorAssembler
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

import datetime
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
%matplotlib inline
# create a Spark session
spark = SparkSession \
    .builder \
    .appName("Sparkify Capstone Project") \
    .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>df = spark.read.json('mini_sparkify_event_data.json')
df.take(5)<jupyter_output><empty_output><jupyter_text>Data Understanding  ChurnChurn is defined as a Cancellation Confirmation action.We can also look at Downgrade event, which is when a user downgrades their subscription from Paid level to Free level.<jupyter_code>df.count()
df.printSchema()
df.select('userId').dropDuplicates().sort('userId').show(10)<jupyter_output>+------+
|userId|
+------+
|      |
|    10|
|   100|
|100001|
|100002|
|100003|
|100004|
|100005|
|100006|
|100007|
+------+
only showing top 10 rows<jupyter_text>All possible user actions<jupyter_code>df.select(col('page')).dropDuplicates().show(25)<jupyter_output>+--------------------+
|                page|
+--------------------+
|              Cancel|
|    Submit Downgrade|
|         Thumbs Down|
|                Home|
|           Downgrade|
|         Roll Advert|
|              Logout|
|       Save Settings|
|Cancellation Conf...|
|               About|
| Submit Registration|
|            Settings|
|               Login|
|            Register|
|     Add to Playlist|
|          Add Friend|
|            NextSong|
|           Thumbs Up|
|                Help|
|             Upgrade|
|               Error|
|      Submit Upgrade|
+--------------------+<jupyter_text>Actions without any user id are where user has not registered yet or is looking to login.<jupyter_code>blank_pages = df.filter(df.userId == '') \
    .select(col('page') \
    .alias('blank_pages')) \
    .dropDuplicates()
blank_pages.show()
df.select(['itemInSession', 'userId', 'sessionId']).groupBy(['userId', 'sessionId']).agg({'itemInSession': 'avg'}).sort('userId').collect()
df.select('sessionId').dropDuplicates().sort('sessionId').show(10)
df = df.dropna(how = 'any', subset = ['userId', 'sessionId'])
df.count()
df = df[df.userId != ""]
df.count()
df.where(col("ts").isNull())<jupyter_output><empty_output><jupyter_text>Convert timestamps to useable datetime formats<jupyter_code>new_ts = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).strftime("%m-%d-%Y %H:%M:%S"))
df = df.withColumn('updated_registration', new_ts('registration'))
df = df.withColumn('updated_ts', new_ts('ts'))
df.take(5)
# Checking Downgrade event
flag_downgrade_event = udf(lambda x: 1 if x == "Submit Downgrade" else 0, IntegerType())
df = df.withColumn("downgrade_event", flag_downgrade_event("page"))
windowval = Window.partitionBy('userId')
df = df.withColumn('downgrade', Fmax('downgrade_event').over(windowval))
# Checking Cancellation event
churn_value = udf(lambda x: 1 if x == 'Cancellation Confirmation' else 0, IntegerType())
df = df.withColumn("churn_value", churn_value("page"))

churn_window = Window.partitionBy('UserId')
df = df.withColumn("user_churn", Fmax('churn_value').over(churn_window))
df_pd = df.toPandas()
df_pd.head().T
df_pd.describe()
(df_pd.groupby(['userId', 'level', 'sessionId'])['itemInSession'].mean()).groupby(['userId', 'level']).mean().reset_index()
churn_df = df_pd.groupby('user_churn').userId.nunique().reset_index()
sns.barplot(churn_df.user_churn, churn_df.userId)
plt.title('Distribution of Users by churn')<jupyter_output><empty_output><jupyter_text>**Highly Imbalanced Dataset**<jupyter_code>sns.countplot(df_pd.drop_duplicates('userId').level)
plt.title('Distribution of users by Subscription level')<jupyter_output><empty_output><jupyter_text>More free users than paid users<jupyter_code>df_pd.groupby('gender').userId.nunique()<jupyter_output><empty_output><jupyter_text>Dataset is almost balanced wrt gender of users<jupyter_code>churn_by_gender = df_pd.groupby(['user_churn', 'gender']).userId.nunique().reset_index()
sns.barplot(churn_by_gender.user_churn, churn_by_gender.userId, hue=churn_by_gender.gender)
plt.title('Distribution of churned users by gender')
churn_by_level = df_pd.groupby(['level', 'user_churn']).userId.nunique()
churn_pcts = churn_by_level.groupby(level=0).apply(lambda x:
                                                 100 * x / float(x.sum())).reset_index()
sns.barplot(churn_pcts.user_churn, churn_pcts.userId, hue=churn_pcts.level)
plt.title('Percent of churned users by Subscription level')<jupyter_output><empty_output><jupyter_text>Churn is almost same for free and paid subscription users wrt churn<jupyter_code>num_songs = df_pd[df_pd.page=='NextSong'].groupby(['user_churn', 'userId', 'sessionId', 'level']).size()\
                             .groupby(['user_churn', 'userId', 'level']).mean().reset_index()
sns.boxplot(num_songs.user_churn, num_songs[0], hue=num_songs.level)
plt.title('Number of songs per session by Subscription level')<jupyter_output><empty_output><jupyter_text>Paid users have much higher usage per session compared to free users.<jupyter_code>df_pd.artist.value_counts().head(15)<jupyter_output><empty_output><jupyter_text>Data Preparation We aggregate data at user-level from interactions to identify features for churn prediction.<jupyter_code>total_songs = df.select('userId', 'song').groupBy('userId').count().withColumnRenamed('count', 'total_songs')
total_songs.show(10)
total_length = df.select('userId', 'length').groupBy('userId').sum().withColumnRenamed('sum(length)', 'total_length')
total_length.show(10)
thumbs_up = df.select('userId', 'page').where(df.page == 'Thumbs Up').groupBy('userId').count().withColumnRenamed('count', 'thumbs_up')
thumbs_down = df.select('userId', 'page').where(df.page == 'Thumbs Down').groupBy('userId').count().withColumnRenamed('count', 'thumbs_down')
user_lifetime = df.select('userId', 'registration', 'ts').withColumn('tot_lifetime', (df.ts - df.registration))\
    .groupBy('userId').agg({'tot_lifetime' : 'max'})\
    .withColumnRenamed('max(tot_lifetime)', 'tot_lifetime')\
    .select('userId', (col('tot_lifetime')/1000/3600/24).alias('user_lifetime'))
user_lifetime.show(10)
friends_added = df.select('userId', 'page').where(df.page == 'Add Friend').groupBy('userId').count().withColumnRenamed('count', 'friends_added')
friends_added.show(10)
songs_per_session = df.where('page == "NextSong"').groupby(['userId', 'sessionId']).count().groupby('userId')\
          .agg({'count' : 'avg'}).withColumnRenamed('avg(count)', 'songs_per_session')
songs_per_session.show(10)
total_artists = df.filter(df.page == "NextSong").select("userId", "artist").dropDuplicates().groupby("userId").count()\
         .withColumnRenamed("count", "total_artists")
total_artists.show(10)
gender = df.select('userId', 'gender').dropDuplicates().replace(['F', 'M'], ['1', '0'], 'gender').select('userId', col('gender').cast('int'))
gender.show(10)
playlist_added = df.select('userID','page').where(df.page == 'Add to Playlist').groupBy('userID').count().withColumnRenamed('count', 'playlist_added')
playlist_added.show(10)
downgrade = df.select('userId', 'downgrade').dropDuplicates().select('userId', col('downgrade'))
gender.show(10)
user_churn = df.select('userId', col('user_churn')).dropDuplicates()
user_churn.show()
final_df = total_songs.join(total_length, 'userId', 'inner').join(thumbs_up, 'userId', 'inner')\
                .join(thumbs_down, 'userId', 'inner').join(user_lifetime, 'userId', 'inner').join(songs_per_session, 'userId', 'inner')\
                .join(friends_added, 'userId', 'inner').join(playlist_added, 'userId', 'inner').join(gender, 'userId', 'inner')\
                .join(total_artists, 'userId', 'inner').join(downgrade, 'userId', 'inner')\
                .join(user_churn, 'userId', 'inner').drop('userId')

final_df.show(10)<jupyter_output>+-----------+------------------+---------+-----------+------------------+------------------+-------------+--------------+------+-------------+---------+----------+
|total_songs|      total_length|thumbs_up|thumbs_down|     user_lifetime| songs_per_session|friends_added|playlist_added|gender|total_artists|downgrade|user_churn|
+-----------+------------------+---------+-----------+------------------+------------------+-------------+--------------+------+-------------+---------+----------+
|        381| 66940.89735000003|       17|          5|  55.6436574074074|39.285714285714285|            4|             7|     1|          252|        0|         0|
|        474| 94008.87593999993|       21|          6| 70.07462962962963|              64.5|            4|             8|     0|          339|        0|         0|
|       4825|1012312.0927899999|      171|         41|131.55591435185184|145.67857142857142|           74|           118|     1|         2232|        0|         0|
|       2464| 52[...]<jupyter_text>Modeling We convert features to numeric features and then scale them.Dataset is split into train-validation-test set in 70:15:15 ratio.Since the dataset is imbalanced, we use F1 as metric for evaluating the models, rather than accuracy.After which fit different models on training dataset and check the performance on the validation set. Once we identify the best model, we perform hyperparameter tuning, after which we identify the best model.The final model is used to make predictions for the test set.<jupyter_code>assmblr = VectorAssembler(inputCols = ['total_songs', 'total_length', 'thumbs_up', 'thumbs_down', 'user_lifetime', 
                                       'songs_per_session', 'friends_added', 'playlist_added', 'gender', 'total_artists', 
                                       'downgrade'],
                          outputCol = "VectFeatures")
final_df = assmblr.transform(final_df)
final_df.show(5)
scaler = StandardScaler(inputCol = 'VectFeatures', outputCol = "scaled_features", withStd = True)
scaler_model = scaler.fit(final_df)
final_df = scaler_model.transform(final_df)
final_df.show(5)
data = final_df.select(final_df.user_churn.alias("label"), final_df.scaled_features.alias("features"))
data.show(5)<jupyter_output>+-----+--------------------+
|label|            features|
+-----+--------------------+
|    0|[0.28303433540817...|
|    0|[0.35212145664954...|
|    0|[3.58435870956552...|
|    1|[1.83043727676051...|
|    0|[0.14931732655392...|
+-----+--------------------+
only showing top 5 rows<jupyter_text>Train-test-validation split<jupyter_code>train, rest = data.randomSplit([0.7, 0.3])
val, test = rest.randomSplit([0.5, 0.5])
train.select('label').agg({'label': 'sum'}).collect()
train.count()<jupyter_output><empty_output><jupyter_text>**Baseline Model**Predicts 0 for all records<jupyter_code>dummy_classifier = val.withColumn('prediction', lit(0.0))
print('Dummy Classifier metrics:')
evaluator = MulticlassClassificationEvaluator(predictionCol='prediction')
print('Accuracy: {}'.format(evaluator.evaluate(dummy_classifier, {evaluator.metricName: "accuracy"})))
print('F-1 Score:{}'.format(evaluator.evaluate(dummy_classifier, {evaluator.metricName: "f1"})))
print('Recall Score:{}'.format(evaluator.evaluate(dummy_classifier, {evaluator.metricName: "weightedRecall"})))
lr = LogisticRegression(maxIter=20)
rf = RandomForestClassifier()
svc = LinearSVC(maxIter=20)
gb = GBTClassifier(maxIter=20)
model_lr = lr.fit(train)
results_lr = model_lr.transform(val)

print('Logistic Regression metrics:')
evaluator = MulticlassClassificationEvaluator(predictionCol='prediction')
print('Accuracy: {}'.format(evaluator.evaluate(results_lr, {evaluator.metricName: "accuracy"})))
print('F-1 Score:{}'.format(evaluator.evaluate(results_lr, {evaluator.metricName: "f1"})))
model_rf = rf.fit(train)
results_rf = model_rf.transform(val)

print('Random Forest metrics:')
evaluator = MulticlassClassificationEvaluator(predictionCol='prediction')
print('Accuracy: {}'.format(evaluator.evaluate(results_rf, {evaluator.metricName: "accuracy"})))
print('F-1 Score:{}'.format(evaluator.evaluate(results_rf, {evaluator.metricName: "f1"})))
model_svc = svc.fit(train)
results_svc = model_svc.transform(val)

print('Linear SVC metrics:')
evaluator = MulticlassClassificationEvaluator(predictionCol='prediction')
print('Accuracy: {}'.format(evaluator.evaluate(results_svc, {evaluator.metricName: "accuracy"})))
print('F-1 Score:{}'.format(evaluator.evaluate(results_svc, {evaluator.metricName: "f1"})))
gb = GBTClassifier(maxIter=15)
model_gb = gb.fit(train)
results_gb = model_gb.transform(val)

print('GB Tree metrics:')
evaluator = MulticlassClassificationEvaluator(predictionCol='prediction')
print('Accuracy: {}'.format(evaluator.evaluate(results_gb, {evaluator.metricName: "accuracy"})))
print('F-1 Score:{}'.format(evaluator.evaluate(results_gb, {evaluator.metricName: "f1"})))<jupyter_output>GB Tree metrics:
Accuracy: 0.8666666666666667
F-1 Score:0.8499999999999999<jupyter_text>Hyper-parameter TuningUsing F1 score, best model is Random Forest model. We tune the model.<jupyter_code># Model Tuning
rf_tune = RandomForestClassifier()
f1_evaluation = MulticlassClassificationEvaluator(metricName = 'f1')
param_grid_rf_tune = ParamGridBuilder().addGrid(rf_tune.numTrees, [15, 20, 25, 30, 35, 40])\
                 .addGrid(rf_tune.maxDepth, [5, 10, 15]).build()

cross_rf_tune = CrossValidator(estimator = rf_tune, estimatorParamMaps = param_grid_rf_tune, evaluator = f1_evaluation, numFolds = 3)
model_rf_tune = cross_rf_tune.fit(train)
results_rf_tune = model_rf_tune.transform(val)

print('Random Tree Tuned metrics:')
evaluator = MulticlassClassificationEvaluator(predictionCol='prediction')
print('Accuracy: {}'.format(evaluator.evaluate(results_rf_tune, {evaluator.metricName: "accuracy"})))
print('F-1 Score:{}'.format(evaluator.evaluate(results_rf_tune, {evaluator.metricName: "f1"})))<jupyter_output>Random Tree Tuned metrics:
Accuracy: 0.8333333333333334
F-1 Score:0.8039579468150897<jupyter_text>**Making Predictions for Test set using Best Model**<jupyter_code>results_test_rf = model_rf.transform(test)

print('Test Set metrics:')
evaluator = MulticlassClassificationEvaluator(predictionCol='prediction')
print('Accuracy: {}'.format(evaluator.evaluate(results_test_rf, {evaluator.metricName: "accuracy"})))
print('F-1 Score:{}'.format(evaluator.evaluate(results_test_rf, {evaluator.metricName: "f1"})))<jupyter_output>Test Set metrics:
Accuracy: 0.8095238095238095
F-1 Score:0.7913165266106442<jupyter_text>**Feature Importance**<jupyter_code>model_rf.featureImportances.values
df_feat_imp = pd.DataFrame([model_rf.featureImportances.values], columns=['total_songs','total_length','thumbs_up','thumbs_down',
                                                            'user_lifetime','songs_per_session','friends_added','playlist_added',
                                                            'gender','total_artists','downgrade'], index=['Feature_Importance']).T.sort_values('Feature_Importance')
import numpy as np
plt.figure(figsize=(10,5))
sns.barplot(df_feat_imp.index, df_feat_imp.Feature_Importance)
plt.xticks(rotation='vertical')
plt.title("Feature Importance of RandomForest model")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.16527506365843547, 'fraction_numerical': 0.03292473500325694, 'mean_word_length': 4.562582345191041, 'pattern_counts': {'":': 0, '<': 66, '<?xml version=': 0, '>': 66, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 3, 'word_list_counts': {'cursed_substrings.json': 11, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '18', 'dir': 'jupyter-structured-clean-dedup', 'id': '172074', 'license': 'MIT', 'n_tokens_mistral': 5794, 'n_tokens_neox': 5446, 'n_words': 1056, 'path': '6_Sparkify_Capstone/Sparkify.ipynb', 'repo_name': 'Sakchhi/Udacity_DS_Nanodegree', 'size': '16887'}