<jupyter_start><jupyter_text>Customer churn predictionThis notebook uses a subset (128MB) of the full dataset available (12GB) for the convinence of manipulating data and build machine learning model locally.<jupyter_code># import libraries
import datetime
import time

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
%matplotlib inline


from pyspark.sql import SparkSession, Window
from pyspark.sql.functions import col, isnan, count,countDistinct, udf, min, max, avg, desc, when, length, datediff
from pyspark.sql.functions import sum as Fsum
from pyspark.sql.types import IntegerType

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

# 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.   1. Load and overview of the data<jupyter_code># Load and read data as dataframe
path = "mini_sparkify_event_data.json"
df = spark.read.json(path)
df.printSchema()
num_rows = df.count()
num_cols = len(df.columns)
print('The data set contains {} rows and {} columns'.format(num_rows, num_cols))
# Show the fist row of the dataframe
df.show(n=1)
# Show all the columns
df.columns<jupyter_output><empty_output><jupyter_text>2. Check missing values and clean data<jupyter_code>def check_missing_values(df):
    """Check the number of missing values in each columns and calculate the missing rate
    
    Arguments:
        df {dataframe} -- spark dataframe
        
    Return: None    
    """
    print('Column name         null count   null rate\n')
    for c in df.columns:
        null_count = df.filter(df[c].isNull() | (df[c] == '') ).count() # Check whether each column has null value or empty string
        null_rate = null_count/df.count() 
        print('{:15} | {:10} | {:10}'.format(c, null_count, null_rate))
check_missing_values(df)<jupyter_output>Column name         null count   null rate

artist          |      58392 | 0.20381151832460734
auth            |          0 |        0.0
firstName       |       8346 | 0.029130890052356022
gender          |       8346 | 0.029130890052356022
itemInSession   |          0 |        0.0
lastName        |       8346 | 0.029130890052356022
length          |      58392 | 0.20381151832460734
level           |          0 |        0.0
location        |       8346 | 0.029130890052356022
method          |          0 |        0.0
page            |          0 |        0.0
registration    |       8346 | 0.029130890052356022
sessionId       |          0 |        0.0
song            |      58392 | 0.20381151832460734
status          |          0 |        0.0
ts              |          0 |        0.0
userAgent       |       8346 | 0.029130890052356022
userId          |       8346 | 0.029130890052356022<jupyter_text>It is shown above that columns such as 'firstName', 'gender', 'lastName', 'location', 'registration' and 'userAgent' all have 8346 missing values as 'userId', because these information are all 'userId' related. Since our purpose in this project is to analyze user behavior, it would be safe to drop those rows that have no user information.<jupyter_code># Drop rows with missing user id
df_dropId = df.filter((df['userId'] != '' ) )
# Total rows left
df_dropId.count()<jupyter_output><empty_output><jupyter_text>Another case is that 3 song-related columns 'artist', 'length' and 'song' have the same number of missing values. Let's check their interaction with 'page' column to get some idea.<jupyter_code>df.groupby('page').agg({'artist':'count', 'length':'count', 'song':'count'}).show(25)<jupyter_output>+--------------------+-----------+-------------+-------------+
|                page|count(song)|count(artist)|count(length)|
+--------------------+-----------+-------------+-------------+
|              Cancel|          0|            0|            0|
|    Submit Downgrade|          0|            0|            0|
|         Thumbs Down|          0|            0|            0|
|                Home|          0|            0|            0|
|           Downgrade|          0|            0|            0|
|         Roll Advert|          0|            0|            0|
|              Logout|          0|            0|            0|
|       Save Settings|          0|            0|            0|
|Cancellation Conf...|          0|            0|            0|
|               About|          0|            0|            0|
| Submit Registration|          0|            0|            0|
|            Settings|          0|            0|            0|
|               Login|          0|            0|       [...]<jupyter_text>The three song related columns exclusively relate with 'NextSong' in page column. We will explore it further in feature engineering section.<jupyter_code># df_drop = df_dropId.drop('artist', 'song')<jupyter_output><empty_output><jupyter_text>Now, with missing user ids removed, let's see how many users in total are registered.<jupyter_code># Check number of unique users
n_users = df_dropId.select('userId').dropDuplicates().count()
print('There are {} unique users in the dataset.'.format(n_users))<jupyter_output>There are 225 unique users in the dataset.<jupyter_text>Since user ids are unique values to idendify users, we don't need extra 'firstName' and 'lastName' columns.<jupyter_code>df_drop = df_dropId.drop('firstName','lastName')
# Take a look at numerical columns
df_drop.select('itemInSession','registration','length','sessionId', 'status','ts').describe().show()

# Take a look at categorical columns
# count and plot occurences in each category in each column
cat_cols = ['auth', 'gender', 'level', 'page']
for c in cat_cols:
    grouped = df_drop.groupby(c).count().toPandas()
    grouped.plot.bar(x=c)<jupyter_output><empty_output><jupyter_text>The 'status', 'method' and 'userAgent'column contains network related records, can also be dropped.<jupyter_code>df_drop = df_drop.drop('status','method','userAgent')
# Show cleaned dataset
# df_drop.show(5)<jupyter_output><empty_output><jupyter_text>Above we cleaned the dataframe by dropping  rows with mising user id and columns that are not relevant. The next step is to define churn and perform exploratory data analysis.    Exploratory Data Analysis 1. Define Churn About a detailed explanation about churn and churn rate, here is a [link](https://www.profitwell.com/blog/the-complete-saas-guide-to-calculating-churn-rate-and-keeping-it-simple) that I read and think is helpful. According to the link and my own understanding, in our dataset we can define churn using column "Cancelation Confirmation".Let's take a look at the data to identify user churn and answer some questions.  2. Data Exploration<jupyter_code># Data overview
df_drop.show(5)
# Format registration time and user behavior time
format_time = udf(lambda x: datetime.datetime.fromtimestamp(x / 1000.0).strftime("%Y-%m-%d %H:%M:%S"))
df_format = df_drop.withColumn('time',format_time(df_drop.ts))
df_format = df_format.withColumn('registrationTime', format_time(df_format.registration))
df_format = df_format.drop('ts','registration')
df_format.show(5)<jupyter_output>+----------------+---------+------+-------------+---------+-----+--------------------+--------+---------+--------------------+------+-------------------+-------------------+
|          artist|     auth|gender|itemInSession|   length|level|            location|    page|sessionId|                song|userId|               time|   registrationTime|
+----------------+---------+------+-------------+---------+-----+--------------------+--------+---------+--------------------+------+-------------------+-------------------+
|  <NAME>|Logged In|     M|           50|277.89016| paid|     Bakersfield, CA|NextSong|       29|           Rockpools|    30|2018-09-30 19:01:57|2018-09-28 17:22:42|
|Five Iron Frenzy|Logged In|     M|           79|236.09424| free|Boston-Cambridge-...|NextSong|        8|              Canada|     9|2018-09-30 19:03:00|2018-09-30 13:20:30|
|    <NAME>|Logged In|     M|           51| 282.8273| paid|     Bakersfield, CA|NextSong|       29|   Time For Miracles|    [...]<jupyter_text>Question 1: How many users canceled their accounts? What's the proportion to all users?<jupyter_code>cancel_user = df_format.filter(df_format['page'] == 'Cancellation Confirmation').select('userId').dropDuplicates().count()
cancel_rate = cancel_user/ n_users * 100
print('{} users out of {} cancelled their account. The cancel rate is {:.2f}%'.format(cancel_user, n_users, cancel_rate))<jupyter_output>52 users out of 225 cancelled their account. The cancel rate is 23.11%<jupyter_text>Question 2: How many users with free account canceled?<jupyter_code># free_cancel_user = df_drop.filter((df_drop['level'] == 'free') & (df_drop['page'] == 'Cancellation Confirmation'))
# free_cancel_user = df_format.filter((df_drop['level'] == 'free' ) & (df_format['page'] == 'Cancellation Confirmation')).select('userId','page')
# free_cancel_user.show(5)
free_cancel_user = df_format.filter((df_format['level'] == 'free' ) & (df_format['page'] == 'Cancellation Confirmation')).dropDuplicates().count()
free_cancel_rate = free_cancel_user/n_users * 100
print('{} free users out of {} cancelled their account. The cancel rate is {:.2f}%'.format(free_cancel_user, n_users, free_cancel_rate))<jupyter_output>21 free users out of 225 cancelled their account. The cancel rate is 9.33%<jupyter_text>Question 3: How many users with paid account canceled?<jupyter_code>paid_cancel_user = df_format.filter((df_format['level'] == 'paid' ) & (df_format['page'] == 'Cancellation Confirmation')).dropDuplicates().count()
paid_cancel_rate = paid_cancel_user/n_users * 100
print('{} paid users out of {} cancelled their account. The cancel rate is {:.2f}%'.format(paid_cancel_user, n_users, paid_cancel_rate))

# Plot to see the distribution between paid and free users
level_df = df_format.filter('page == "Cancellation Confirmation"').groupby('level').count().toPandas()
sns.barplot(data = level_df, x = 'level', y = 'count', color = sns.color_palette()[0])<jupyter_output><empty_output><jupyter_text>3. Create churn label to identify users who churned or not<jupyter_code># Create a boolean function to check if a user canceled or not
flag_cancel_event = udf(lambda x: 1 if x == 'Cancellation Confirmation' else 0, IntegerType())
df_format = df_format.withColumn('cancel_event', flag_cancel_event('page'))
# df_format.show(5)
# create churn label
flag_churn_event = udf(lambda x: 1 if x == 'Cancellation Confirmation' else 0, IntegerType())
df_format = df_format.withColumn('churn_event', flag_churn_event('page'))

# label user who churned
windowval = Window.partitionBy('userId')
df_format = df_format.withColumn('churned', max('churn_event').over(windowval))
df_format.show(5)<jupyter_output>+--------------------+---------+------+-------------+---------+-----+--------------------+---------+---------+--------------------+------+-------------------+-------------------+------------+-----------+-------+
|              artist|     auth|gender|itemInSession|   length|level|            location|     page|sessionId|                song|userId|               time|   registrationTime|cancel_event|churn_event|churned|
+--------------------+---------+------+-------------+---------+-----+--------------------+---------+---------+--------------------+------+-------------------+-------------------+------------+-----------+-------+
|Sleeping With Sirens|Logged In|     F|            0|202.97098| free|Bridgeport-Stamfo...| NextSong|       31|Capt<NAME>...|100010|2018-10-08 07:58:54|2018-09-26 21:45:40|           0|          0|      0|
|<NAME>...|Logged In|     F|            1|196.54485| free|Bridgeport-Stamfo...| NextSong|       31|Beautiful_ Beauti...|100010|2018-10-08 08:[...]<jupyter_text>Question 1: Churn happened most likely at what time of the day?<jupyter_code># format timestamp to hours and creat a new column
get_hour = udf(lambda x: datetime.datetime.fromtimestamp(x/1000.0).hour, IntegerType())
df_hour = df.withColumn('hour',get_hour(df.ts))

# format timestamp to days of a week and creat a new column
get_weekday = udf(lambda x: datetime.datetime.fromtimestamp(x/1000.0).weekday(), IntegerType())
df_weekday = df.withColumn('weekday',get_weekday(df.ts))



# format timestamp to days and creat a new column
get_day = udf(lambda x: datetime.datetime.fromtimestamp(x/1000.0).day, IntegerType())
df_day = df.withColumn('day',get_day(df.ts))

# Count numbers of users who churned by hours of the day and plot the result
cancel_hour = df_hour.filter(df_hour.page == 'Cancellation Confirmation').select(['userId','page','hour']).toPandas()
user_count_cancel = cancel_hour.groupby('hour').userId.count()
user_count_cancel= user_count_cancel.reset_index(name='userCount').sort_values(by='hour')
sns.barplot(data=user_count_cancel, x=user_count_cancel['hour'], y=user_count_cancel['userCount'], color=sns.color_palette()[0])
plt.xlabel('Hour of a day')
plt.ylabel('Number of churned users')
# Count numbers of users who churned by days of the week and plot the result
cancel_weekday = df_weekday.filter(df_weekday.page == 'Cancellation Confirmation').select(['userId','page','weekday']).toPandas()
user_count_weekday = cancel_weekday.groupby('weekday').userId.count()
user_count_weekday= user_count_weekday.reset_index(name='userCount').sort_values(by='weekday')
sns.barplot(data=user_count_weekday, x=user_count_weekday['weekday'], y=user_count_cancel['userCount'], color=sns.color_palette()[0])
plt.xlabel('Days of a week')
plt.ylabel('Number of churned users')<jupyter_output><empty_output><jupyter_text>It shows that more people churned on Tuesday<jupyter_code># Count numbers of users who churned by days of the week and plot the result
cancel_day = df_day.filter(df_day.page == 'Cancellation Confirmation').select(['userId','page','day']).toPandas()
user_count_day = cancel_day.groupby('day').userId.count()
user_count_day= user_count_day.reset_index(name='userCount').sort_values(by='day')
sns.barplot(data=user_count_day, x=user_count_day['day'], y=user_count_day['userCount'], color=sns.color_palette()[0])
plt.xlabel('Days of a month')
plt.ylabel('Number of churned users')<jupyter_output><empty_output><jupyter_text>The above bar chart shows that people tend to Churn at 10am, 1pm and 2am.    Question 2: Is there any difference in genders in terms of churn?<jupyter_code>gender_churn = df_format.drop_duplicates(['userId', 'gender']).groupby(['churned', 'gender']).count().sort("churned").toPandas()
gender_churn.head()
sns.barplot(x = 'churned', y = 'count', data = gender_churn, hue = 'gender')<jupyter_output><empty_output><jupyter_text>We can see that there are more men users than women users in both churned and not churned category.   Question 3: Is there any difference between churned and not churned users with respect to number of songs played per session?<jupyter_code>total_songs_df = df_format.where('page == "NextSong"').groupby(['churned', 'userId', 'sessionId']).count()\
    .groupby(['churned', 'userId']).agg({'count': 'avg'})\
    .withColumnRenamed('avg(count)', 'songsPerSession').toPandas()
total_songs_df.head()
# total_songs_df = df_format.filter(df_format.page == 'NextSong').select('churned','userId','sessionId').toPandas()
# total_songs_df.head()
# total_songs_df.groupby('churned').agg({'sessionId':ave})
sns.barplot(x = 'churned', y = 'songsPerSession', data = total_songs_df, hue = 'churned')<jupyter_output><empty_output><jupyter_text>Feature EngineeringSince we now familiar with the data, the next stept is to build out the features I find promising to train the 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 necessary  Feature 1:  Total number of songs per user has listened<jupyter_code>feat1 = df_format.select('userId','song').groupby('userId').count().withColumnRenamed('count','totalSong')
feat1.show(5)
feat1.count()<jupyter_output><empty_output><jupyter_text>Feature 2: User lifetime<jupyter_code>feat2 = df_format.select('userId','registrationTime','time')\
    .withColumn('totalLifetime',datediff(df_format.time, df_format.registrationTime))\
    .groupby('userId')\
    .agg({'totalLifetime':'max'}).withColumnRenamed('max(totalLifetime)','totalLifetime')\
                
feat2.show(5)<jupyter_output>+------+-------------+
|userId|totalLifetime|
+------+-------------+
|100010|           56|
|200002|           70|
|   125|           71|
|    51|           20|
|   124|          132|
+------+-------------+
only showing top 5 rows<jupyter_text>Feature 3: Average number of  songs listened per session<jupyter_code>feat3 = df_format.filter(df.page == 'NextSong') \
    .groupby(['userId', 'sessionId']) \
    .count() \
    .groupby(['userId']) \
    .agg({'count':'avg'}) \
    .withColumnRenamed('avg(count)', 'avgSongsPerSession')
feat3.show(5)
feat3.count()<jupyter_output><empty_output><jupyter_text>Feature 4: Number of thumbs up<jupyter_code>feat4 = df_format.filter(df.page == 'Thumbs Up').select('userId').groupby('userId').count().withColumnRenamed('count','totalThumbsup')
feat4.show(5)
feat4.count()<jupyter_output><empty_output><jupyter_text>Feature 5: Number of thumbs down<jupyter_code>feat5 = df_format.filter(df.page == 'Thumbs Down').select('userId').groupby('userId').count().withColumnRenamed('count','totalThumbsdown')
feat5.show(5)
feat5.count()<jupyter_output><empty_output><jupyter_text>Feature 6:  Total number of adding friends<jupyter_code>feat6 = df_format.filter(df.page == 'Add Friend').select('userId').groupby('userId').count().withColumnRenamed('count','totalAddFriend')
feat6.show(5)
feat6.count()<jupyter_output><empty_output><jupyter_text>Feature 7: Total number of adding to playlist<jupyter_code>feat7 = df_format.filter(df.page == 'Add to Playlist').select('userId').groupby('userId').count().withColumnRenamed('count','totalAddList')
feat7.show(5)

feat7.count()<jupyter_output><empty_output><jupyter_text>Feature 8: Total length of listening to songs per user<jupyter_code>feat8 = df_format.select('userId','length').groupby('userId').sum().withColumnRenamed('sum(length)', 'totalListenLength')
feat8.show(5)
feat8.count()<jupyter_output><empty_output><jupyter_text>Feature 9: Total number of interactions per user<jupyter_code>feat9 = df_format.groupby('userId').count().withColumnRenamed('count','totalInteraction')
feat9.show(5)
feat9.count()<jupyter_output><empty_output><jupyter_text>Feature 10: Whether a user downgraded or not<jupyter_code># create downgrade flag
flag_downgrade_event = udf(lambda x: 1 if x == 'Submit Downgrade' else 0, IntegerType())
df_format = df_format.withColumn('downgrade_event', flag_downgrade_event('page'))

# label user who downgraded
windowvald = Window.partitionBy('userId')
df_format = df_format.withColumn('downgraded', max('downgrade_event').over(windowvald))

feat10 = df_format.select('userId','downgraded').dropDuplicates()
feat10.show(10)
feat10.count()<jupyter_output><empty_output><jupyter_text>Feature 11:  Whether a user paid or not<jupyter_code># create paid flag
flag_paid_event = udf(lambda x: 1 if x == 'paid' else 0, IntegerType())
df_format = df_format.withColumn('paid_event', flag_paid_event('level'))

# label user who paid
windowvalp = Window.partitionBy('userId')
df_format = df_format.withColumn('paid', max('paid_event').over(windowvalp))

feat11 = df_format.select('userId','paid').dropDuplicates()
feat11.show(5)
feat11.count()<jupyter_output><empty_output><jupyter_text>Label for prediction<jupyter_code>label = df_format.select('userId',df_format.churned.alias('label')).dropDuplicates()
label.show(5)
label.count()<jupyter_output><empty_output><jupyter_text>Wrap up all the features together We can see from the above information some of the users have no information with respect to features 'totalThumbsUp', 'totalThumbsDown', 'totalAddFriend' and 'totalAddList', we will impute these value as zeros.<jupyter_code>model_df = feat1.join(feat2,'userId','outer') \
    .join(feat3,'userId','outer') \
    .join(feat4,'userId','outer') \
    .join(feat5,'userId','outer') \
    .join(feat6,'userId','outer') \
    .join(feat7,'userId','outer') \
    .join(feat8,'userId','outer') \
    .join(feat9,'userId','outer') \
    .join(feat10,'userId','outer') \
    .join(feat11,'userId','outer') \
    .join(label,'userId','outer') \
    .drop('userId') \
    .fillna(0)

model_df.show(5)
model_df.count()
model_df.write.mode('overwrite').save('model.json')<jupyter_output><empty_output><jupyter_text>ModelingIn the following section, we will split the full dataset into train, test, and validation sets, test out several of the machine learning methods and evaluate the accuracy of the various models, tuning parameters as necessary. Then we will determine the best model based on test accuracy and report results on the validation set. Since the churned users are a fairly small subset, we will use F1 score as the metric to optimize.<jupyter_code># load data
data = spark.read.load('model.json')
data.show(5)<jupyter_output>+---------+-------------+------------------+-------------+---------------+--------------+------------+------------------+----------------+----------+----+-----+
|totalSong|totalLifetime|avgSongsPerSession|totalThumbsup|totalThumbsdown|totalAddFriend|totalAddList| totalListenLength|totalInteraction|downgraded|paid|label|
+---------+-------------+------------------+-------------+---------------+--------------+------------+------------------+----------------+----------+----+-----+
|     1245|          172|              47.1|           35|             11|            19|          23| 231480.3892899999|            1245|         1|   1|    0|
|     2132|          125|              68.5|           84|             17|            23|          43|441694.46125000034|            2132|         0|   1|    0|
|     1950|           71| 91.33333333333333|           84|              9|            27|          58|412321.50108000037|            1950|         0|   1|    0|
|       48|          100|         [...]<jupyter_text>Vectorize features our dataset has 11 features including 9 numerical features and 2 binary features. We will first vectorize each feature and then build a pipeline for vectorizing, scaling and traing.<jupyter_code>numerical_cols = ['totalSong','totalLifetime','avgSongsPerSession','totalThumbsup','totalThumbsdown',
                  'totalAddFriend','totalAddList','totalListenLength','totalInteraction']
binary_cols = ['downgraded','paid']


# Vectorize and standarize numerical features
num_assembler = VectorAssembler(inputCols=numerical_cols, outputCol='num_vect')
scaler = StandardScaler(inputCol='num_vect', outputCol='num_scaled', withStd= True, withMean=True)

# Add binary features
assembler = VectorAssembler(inputCols = binary_cols + ["num_scaled"], outputCol = "features")<jupyter_output><empty_output><jupyter_text>Train/test split<jupyter_code>train,test = data.randomSplit([0.8,0.2], seed=42)
train.count(),test.count()
# Create a function for training
def train_model(train_data,classifier):
    """ A function to train the model
    
    Arguments:
        train_data {dataframe} -- train data
    """
    start = time.time()
    model = classifier.fit(train)
    end = time.time()
    print('Time spent for training: {} seconds'.format(round(end-start)))
    return model

# Create an function for predicting and evaluating
def predict_and_evaluate(test_data, model, metric='f1'):
    """ A function to evaluate a model's performance
    
    Arguments:
        model {object} -- trained model or pipeline object
        test_data {dataframe} -- test data
    
    Keyword Arguments:
        metric {str} -- name of the metric (default: {'f1'}) 
        
    Returns:
        f1 score
    """
    
    evaluator = MulticlassClassificationEvaluator(metricName = metric)
    predictions = model.transform(test_data)
    
    # calcualte score
    score = evaluator.evaluate(predictions)
    print('The f1 score is: {}'.format(score))    
    return score<jupyter_output><empty_output><jupyter_text>Model selection Since the label we are going to predict is binary (whether or not the user will churn), we are going to choose 3 classification models: logistic Regression, Random Forest classifier and Gradient Boosted tree. First we are going to train these 3 models without hyperparameter tuning, compare the results, and then we will perform Grid search cross validation to fine tune our models. Finally we will pick one model that performed the best.   Logistic Regression<jupyter_code>lr = LogisticRegression()
pipeline_lr = Pipeline(stages=[num_assembler, scaler, assembler, lr])
# Train without tuning: default parameters maxIter=100, regParam=0.0, elasticNetParam=0.0
model_lr = train_model(train, pipeline_lr)
# Predict and show f1 score
score_lr = predict_and_evaluate(test, model_lr)
score_lr<jupyter_output>The f1 score is: 0.7781905594405595<jupyter_text>The F1 score look pretty promising with default parameters of logistic regression. We will perform gridsearch cross validation next, and compare the two results.  Use grid search cross validation<jupyter_code># Logistic Regression grid search
# Parameter combinations
param_grid_lr = ParamGridBuilder()\
    .addGrid(lr.maxIter, [10, 50, 100, 200]) \
    .addGrid(lr.regParam, [0.0, 0.05, 0.1, 0.3]) \
    .addGrid(lr.elasticNetParam, [0.0, 0.1, 0.5, 1.0]) \
    .build()

# Grid search with cross validation
cv_lr = CrossValidator(estimator = pipeline_lr,
                          estimatorParamMaps = param_grid_lr,
                          evaluator = MulticlassClassificationEvaluator(),
                          numFolds = 4)
# train model with best parameters
cv_model_lr = train_model(train, cv_lr)
# Predict and show f1 score
cv_score_lr = predict_and_evaluate(test,cv_model_lr)
cv_score_lr<jupyter_output>The f1 score is: 0.7933448573898011<jupyter_text>By comparing the two scores, we can see that grid search cross validation didn't give as a better score. We will use the score with default parameter to comare with othe models.  Random Forest Classifier<jupyter_code>rf = RandomForestClassifier(labelCol='label', featuresCol='features', numTrees=10)
pipeline_rf = Pipeline(stages = [num_assembler, scaler, assembler, rf])
# Train without tuning: default parameters maxDepth=5, maxBins=32, numTrees=20, impurity='gini'
model_rf = train_model(train,pipeline_rf)

# Predict and show f1 score
score_rf = predict_and_evaluate(test,model_rf)
score_rf<jupyter_output>Time spent for training: 1 seconds
The f1 score is: 0.7612942612942614<jupyter_text>Use grid search cross validation<jupyter_code># Random forest classifier grid search
# Parameter combinations
paramGrid_rf = ParamGridBuilder() \
    .addGrid(rf.numTrees, [10, 20, 40]) \
    .addGrid(rf.maxDepth, [4, 5]) \
    .build()

# Grid search with cross validation
cv_rf = CrossValidator(estimator = pipeline_rf,
                          estimatorParamMaps = paramGrid_rf,
                          evaluator = MulticlassClassificationEvaluator(),
                          numFolds = 4)
# Train model with best prarameters
cv_model_rf = train_model(train,cv_rf)
# Store grid search results in a dataframe
scores_rf = cv_model_rf.avgMetrics
params = [{p.name: v for p, v in m.items()} for m in cv_model_rf.getEstimatorParamMaps()]
params_pd = pd.DataFrame(params)
params_pd['f1_score'] = scores_rf
params_pd
# Predict and show f1 score
cv_score_rf = predict_and_evaluate(test,cv_model_rf)
cv_score_rf<jupyter_output>The f1 score is: 0.8225524475524475<jupyter_text>Gradient Boosted Tree<jupyter_code>gbt = GBTClassifier(labelCol='label',featuresCol='features')
pipeline_gbt = Pipeline(stages = [num_assembler, scaler, assembler, gbt])
# Train data without tuning: default parameters maxDepth=5, maxBins=32, maxIter=20, stepSize=0.1
model_gbt = train_model(train,pipeline_gbt)

# Predict and evaluate
score_gbt = predict_and_evaluate(test,model_gbt)
score_gbt
# Gradient-boosted tree classifier grid search
# Parameter combinations
paramGrid_gbt = ParamGridBuilder() \
.addGrid(gbt.maxDepth, [4, 5]) \
.addGrid(gbt.maxIter, [20, 40]) \
.build()

# Grid search with cross validation
cv_gbt = CrossValidator(estimator = pipeline_gbt,
                          estimatorParamMaps = paramGrid_gbt,
                          evaluator = MulticlassClassificationEvaluator(),
                          numFolds = 4)
# Train with best parameter
cv_model_gbt = train_model(train,cv_gbt)
cv_model_gbt.avgMetrics
# Store grid search results in a dataframe
scores_gbt = cv_model_gbt.avgMetrics
params = [{p.name: v for p, v in m.items()} for m in cv_model_gbt.getEstimatorParamMaps()]
params_pd = pd.DataFrame(params)
params_pd['f1_score'] = scores_gbt
params_pd<jupyter_output><empty_output><jupyter_text>The best combanation of parameter is maxDepth: maxDepth: 5, maxIter: 20<jupyter_code># Predict and evaluate
cv_score_gbt = predict_and_evaluate(test,cv_model_gbt)
cv_score_gbt<jupyter_output>The f1 score is: 0.7604395604395605<jupyter_text>Results The result of each classifier with f1 score as metric is shown below:<jupyter_code>dict = {'classifier': ['Logistic Regression','Random Forest','Gradient Boosted Tree'], 
       'f1': [score_lr, score_rf, score_gbt], 'gridSearchCV_f1': [cv_score_lr, cv_score_rf, cv_score_gbt]}
summary_df = pd.DataFrame(data=dict)
summary_df<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.15306832874714055, 'fraction_numerical': 0.03328581374531711, 'mean_word_length': 3.533894483691568, 'pattern_counts': {'":': 0, '<': 155, '<?xml version=': 0, '>': 155, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 1, 'word_list_counts': {'cursed_substrings.json': 7, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '41', 'dir': 'jupyter-structured-clean-dedup', 'id': '234162', 'license': 'MIT', 'n_tokens_mistral': 9898, 'n_tokens_neox': 9233, 'n_words': 2352, 'path': 'Sparkify.ipynb', 'repo_name': 'gj0706/customer_churn_prediction_with_Spark', 'size': '30195'}