<jupyter_start><jupyter_text>Sparkify Project Workspace<jupyter_code>import os
os.environ["PYSPARK_PYTHON"] = "python3"
os.environ["PYSPARK_DRIVER_PYTHON"] = "python3"
os.environ["JAVA_HOME"] = "/Library/java/JavaVirtualMachines/adoptopenjdk-8.jdk/contents/Home/"
import findspark
findspark.init()
# import libraries
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql import types as T
from pyspark.sql import Window
import datetime

from pyspark.ml import Pipeline
from pyspark.ml.feature import (
    StringIndexer, OneHotEncoderEstimator, MinMaxScaler, VectorAssembler)
from pyspark.ml.classification import (
    DecisionTreeClassifier, RandomForestClassifier, GBTClassifier, LogisticRegression)
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator

from sklearn.metrics import classification_report, confusion_matrix
from sklearn.metrics import roc_curve

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
%matplotlib inline
# Create spark session
spark = SparkSession \
    .builder \
    .appName("Sparkify") \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Load and Clean DatasetLoad and clean the `mini_sparkify_event_data.json` dataset, checking for invalid or missing data - for example, records without user-ids or session-ids.<jupyter_code># Read in sparkify dataset
event_data = "mini_sparkify_event_data.json"
df = spark.read.json(event_data)
df.head()
df.printSchema()
# number of records
df.count()
# check number of missing values in each column
missing_values = [F.count(F.when(F.col(c).isNull(), c)).alias(c)
                  for c in df.columns]
df.select(missing_values).show()
# Users without registration are not valid for churn analysis, and thus will be dropped.
df_valid = df.dropna(subset=['registration'])
# check number of missing values for each column again
missing_values = [F.count(F.when(F.col(c).isNull(), c)).alias(c)
                  for c in df_valid.columns]
df_valid.select(missing_values).show()<jupyter_output>+------+----+---------+------+-------------+--------+------+-----+--------+------+----+------------+---------+-----+------+---+---------+------+
|artist|auth|firstName|gender|itemInSession|lastName|length|level|location|method|page|registration|sessionId| song|status| ts|userAgent|userId|
+------+----+---------+------+-------------+--------+------+-----+--------+------+----+------------+---------+-----+------+---+---------+------+
| 50046|   0|        0|     0|            0|       0| 50046|    0|       0|     0|   0|           0|        0|50046|     0|  0|        0|     0|
+------+----+---------+------+-------------+--------+------+-----+--------+------+----+------------+---------+-----+------+---+---------+------+<jupyter_text>`artist`, `length`, and `song` will be null when a user is visiting pages other than `NextSong`.<jupyter_code>df_valid.filter(F.col('artist').isNotNull()).select(
    'page').dropDuplicates().show()<jupyter_output>+--------+
|    page|
+--------+
|NextSong|
+--------+<jupyter_text>Exploratory Data AnalysisEDA on a small subset of the data with basic manipulations within Spark.<jupyter_code># number of users
num_users = df_valid.select("userId").dropDuplicates().count()
print(num_users)
# number of artists
df_valid.select("artist").dropDuplicates().count()
# auth
df_valid.select("auth").dropDuplicates().show()
# itemInSession
df_valid.select("itemInSession").describe().show()
# length
df_valid.select("length").describe().show()
# histogram for length
df_valid.select("length").toPandas().hist(bins=100);
# level
df.select("level").dropDuplicates().show()
# location
df.select("location").dropDuplicates().show(10, False)
# method
df.select("method").dropDuplicates().show()
# page
df.select("page").dropDuplicates().show(22, False)
# registration
df.select("registration").show(5)
# songs
df.select("song").show(5, False)
# number of songs
df.select("song").drop_duplicates().count()
# status
df.select("status").drop_duplicates().show()
# ts
df.select("ts").show(5)
# userAgent
df_valid.select("userAgent").drop_duplicates().show(10, False)<jupyter_output>+--------------------------------------------------------------------------------------------------------------------------+
|userAgent                                                                                                                 |
+--------------------------------------------------------------------------------------------------------------------------+
|"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"|
|"Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/36.0.1985.143 Safari/537.36"                  |
|Mozilla/5.0 (X11; Ubuntu; Linux i686; rv:31.0) Gecko/20100101 Firefox/31.0                                                |
|"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"|
|"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"|
[...]<jupyter_text>Define ChurnDefine churn events as`Cancellation Confirmation` events.Create`churn_event` and `churn_user` columns.<jupyter_code># define churn event
flag_churn_event = F.udf(
    lambda x: 1 if x == 'Cancellation Confirmation' else 0, T.IntegerType())

# create churn event column
df_valid = df_valid.withColumn('churn_event', flag_churn_event('page'))

# create churn user column
df_valid = df_valid.withColumn('churn_user', F.max(
    'churn_event').over(Window.partitionBy('userId')))
df_valid.groupBy('churn_user').count().show()
44864/(44864 + 233290) * 100<jupyter_output><empty_output><jupyter_text>Explore DataMore exploratory data analysis to observe the behavior of users who stayed vs users who churned. The relations between users' attributes such as gender, level, location, and churn events are explored. We also explore aggregates on these two groups of users, observing how much of a specific action they experienced such as number of songs played or how many times they visited home page per week.<jupyter_code># number of churn and stayed users
num_churn_users = df_valid.filter(df_valid['churn_user'] == '1').select(
    F.countDistinct('userId')).collect()[0][0]
num_stayed_users = num_users - num_churn_users
print("number of users that churn:", num_churn_users)
print("number of users that stayed:", num_stayed_users)
d = {'churns': [num_stayed_users, num_churn_users]}
df_churns = pd.DataFrame(data=d, index=['Stayed', 'Churn'])

ax = df_churns.plot.bar(legend=False)
ax.grid(axis='y')
ax.set_ylabel('Count', fontsize=12)
ax.set_title('')
plt.xticks(rotation=0);
plt.savefig('churns')
# create datetime columns for registration and timestamp
df_valid = df_valid.withColumn(
    'registration_dt', (F.col('registration')/1000).cast(T.TimestampType()))
df_valid = df_valid.withColumn(
    'dt', (F.col('ts')/1000).cast(T.TimestampType()))
# example of churn by `Cancellation Confirmation`
df_valid.select('userId', 'firstName', 'page', 'dt', 'level').where((df_valid['userId'] == 73) & (
    df_valid['page'] == 'Cancellation Confirmation')).show(5, False)
# example of Submit Downgrade`
df_valid.select('userId', 'firstName', 'page', 'dt', 'level').where((df_valid['userId'] == 39) & (
    df_valid['page'] == 'Submit Downgrade')).show(5, False)
# example of churn by `Submit Downgrade` followed by `Cancellation Confirmation`
df_valid.select('userId', 'firstName', 'page', 'dt', 'level').where((df_valid['userId'] == 103) & (
    (df_valid['page'] == 'Submit Downgrade') | (df_valid['page'] == 'Cancellation Confirmation'))).show(5, False)<jupyter_output>+------+---------+-------------------------+-------------------+-----+
|userId|firstName|page                     |dt                 |level|
+------+---------+-------------------------+-------------------+-----+
|103   |Piper    |Submit Downgrade         |2018-10-15 18:56:25|paid |
|103   |Piper    |Cancellation Confirmation|2018-11-04 08:01:31|paid |
+------+---------+-------------------------+-------------------+-----+<jupyter_text>Gender<jupyter_code>num_females = df_valid.filter(df_valid['gender'] == 'F').select(
    F.countDistinct('userId')).collect()[0][0]
num_males = num_users - num_females
print("number of females:", num_females)
print("number of males:", num_males)
df_gender = df_valid.groupby('gender').sum('churn_event').toPandas()
df_gender['count'] = [num_females, num_males]
df_gender['percent churn'] = df_gender['sum(churn_event)'] / \
    df_gender['count']*100
df_gender.columns = ['gender', 'num_churns', 'count', 'percent churn']
df_gender.set_index('gender')
ax = df_gender.plot.bar(x='gender', y='percent churn', legend=False)
ax.grid(axis='y')
ax.set_xlabel('')
ax.set_ylabel('Percent churn', fontsize=12)
ax.set_title('Gender')
plt.xticks(rotation=0);
plt.savefig('gender')<jupyter_output><empty_output><jupyter_text>Item In Session<jupyter_code>df_item_churn = df_valid.filter(df_valid['churn_user'] == 1).select('itemInSession').toPandas()
df_item_stayed = df_valid.filter(df_valid['churn_user'] == 0).select('itemInSession').toPandas()
df_item = pd.concat([df_item_stayed, df_item_churn], axis=1, ignore_index=True)
df_item.describe().transpose()<jupyter_output><empty_output><jupyter_text>Level<jupyter_code>num_free = df_valid.filter(df_valid['level'] == 'free').select(
    F.countDistinct('userId')).collect()[0][0]
num_paid = df_valid.filter(df_valid['level'] == 'paid').select(
    F.countDistinct('userId')).collect()[0][0]
df_level = df_valid.groupby('level').sum('churn_event').toPandas()
df_level['count'] = [num_free, num_paid]
df_level['percent churn'] = df_level['sum(churn_event)'] / \
    df_level['count'] * 100
df_level.columns = ['level', 'num_churns', 'count', 'percent churn']
df_level.set_index('level')
ax = df_level.plot.bar(x='level', y='percent churn', legend=False)
ax.grid(axis='y')
ax.set_ylabel('Percent churn', fontsize=12)
ax.set_xlabel('')
ax.set_title('Level')
plt.xticks(rotation=0);
plt.savefig('level')<jupyter_output><empty_output><jupyter_text>Location<jupyter_code>df_valid.select("location").show(10, False)
# get state from location
get_state = F.udf(lambda x: x[-2:])

# create state column
df_valid = df_valid.withColumn("state", get_state(df_valid['location']))
df_state_counts = df_valid.select(
    'userId', 'state').dropDuplicates().groupby('state').count().toPandas()
ax = df_state_counts.plot.bar(x='state', y='count',
                              figsize=(12, 6))
ax.set_ylabel('number of distinct users', fontsize=12);
df_state_churns = df_valid.groupby('state').sum('churn_event').toPandas()
df_state_churns.columns = ['state', 'num_churns']
ax = df_state_churns.plot.bar(x='state', y='num_churns', figsize=(12, 6), legend=False)
ax.grid(axis='y')
ax.set_xlabel('')
ax.set_ylabel('Churn count', fontsize=16);
#ax.set_title('State', fontsize=18)
plt.xticks(fontsize=14)
plt.yticks(fontsize=14)
plt.tight_layout()

plt.savefig('state')
df_state_churns['percent churn'] = df_state_churns['num_churns'] / \
    df_state_counts['count'] * 100
df_state_churns.plot.bar(x='state', y='percent churn', figsize=(12, 6));<jupyter_output><empty_output><jupyter_text>Registration<jupyter_code># find earliest and latest registration dates
start_reg_date = df_valid.agg({'registration_dt': 'min'}).collect()
end_reg_date = df_valid.agg({'registration_dt': 'max'}).collect()
print('registration start date:', str(start_reg_date[0][0]))
print('registration end date:', str(end_reg_date[0][0]))
# find earliest and latest dates (from timestamp)
start_date = df_valid.agg({'dt': 'min'}).collect()
end_date = df_valid.agg({'dt': 'max'}).collect()
print('start date:', str(start_date[0][0]))
print('end date:', str(end_date[0][0]))
# number of days since registration
df_valid = df_valid.withColumn("days_since_reg", F.max(
    F.datediff('dt', 'registration_dt')).over(Window.partitionBy('userId')))
df_valid.select('days_since_reg').describe().show()
df_valid.select('days_since_reg').toPandas().hist(bins=25)
df_days_stayed = df_valid.filter(
    df_valid['churn_user'] == 0).select('days_since_reg').toPandas()
df_days_churn = df_valid.filter(
    df_valid['churn_user'] == 1).select('days_since_reg').toPandas()
df_reg_days = pd.concat([df_days_stayed, df_days_churn], axis=1, ignore_index=True)
df_reg_days.columns = ['stayed', 'churn']
df_reg_days.describe().transpose()
ax = df_reg_days.boxplot()
ax.set_ylabel('Number of days', fontsize=12)
ax.set_title('Days since registration');
plt.savefig('days_since_registration')<jupyter_output><empty_output><jupyter_text>Timestamp<jupyter_code>get_hour = F.udf(lambda x: datetime.datetime.fromtimestamp(
    x / 1000.0).hour, T.IntegerType())
get_day = F.udf(lambda x: datetime.datetime.fromtimestamp(
    x / 1000.0).day, T.IntegerType())
get_month = F.udf(lambda x: datetime.datetime.fromtimestamp(
    x / 1000.0).month, T.IntegerType())
df_valid = df_valid.withColumn('hour', get_hour(F.col('ts')))\
    .withColumn('day', get_day(F.col('ts')))\
    .withColumn('month', get_month(F.col('ts')))
ax = df_valid.groupBy('hour').sum('churn_event').sort('hour').toPandas(
).plot.bar(x='hour', y='sum(churn_event)', figsize=(12, 6), legend=False)
ax.grid(axis='y')
ax.set_xlabel('Hour', fontsize=16)
ax.set_ylabel('Count', fontsize=16)
ax.set_title('Hourly churn events', fontsize=18)
plt.xticks(fontsize=16)
plt.yticks(fontsize=16)
plt.tight_layout()

plt.savefig('hourly_churn_events')
ax = df_valid.groupBy('day').sum('churn_event').sort('day').toPandas(
).plot.bar(x='day', y='sum(churn_event)', figsize=(12, 6), legend=False)
ax.grid(axis='y')
ax.set_xlabel('Day', fontsize=16)
ax.set_ylabel('Count', fontsize=16)
ax.set_title('Daily churn events', fontsize=18)
plt.xticks(fontsize=16)
plt.yticks(fontsize=16)
plt.tight_layout()

plt.savefig('daily_churn_events')
ax = df_valid.groupBy('month').sum('churn_event').sort(
    'month').toPandas().plot.bar(x='month', y='sum(churn_event)')
ax.grid(axis='y')
ax.set_xlabel('Month', fontsize=12)
ax.set_ylabel('Count', fontsize=12)
ax.set_title('Monthly churn events');
plt.savefig('montly_churn_events')<jupyter_output><empty_output><jupyter_text>Song<jupyter_code># time window to find number of songs played in a week
def days(x): 
    """Return time in seconds for the input time in days."""
    time_in_seconds = x * 86400
    return time_in_seconds


windowval = (
    Window.partitionBy("userId")
    .orderBy(F.col("ts")/1000)
    .rangeBetween(-days(7), 0)
)
# number of songs played in a week
df_valid = df_valid.withColumn("num_songs", F.count("song").over(windowval))
# check
df_valid.select('userId', 'dt', 'song', 'num_songs', 'churn_event',
                'churn_user').where(df_valid['userId'] == 32).show(200)
def make_boxplot(df, col_name):
    """Return boxplot of stayed and churn for the attribute col_name of the dataframe df."""
    df_stayed = df.filter(df['churn_user'] == 0).select(
        col_name).toPandas()
    df_churn = df.filter(df['churn_user'] == 1).select(
        col_name).toPandas()

    df = pd.concat([df_stayed, df_churn], axis=1, ignore_index=True)
    df.columns = ['stayed', 'churn']

    statistics = df.describe().transpose()

    ax = df.boxplot()
    ax.set_title(col_name + ' per week', fontsize=12)
    ax.set_ylabel('Count', fontsize=12)

    return statistics, ax
make_boxplot(df_valid, 'num_songs')<jupyter_output><empty_output><jupyter_text>Artist<jupyter_code># number of distinct artists played in a week
df_valid = df_valid.withColumn("num_artists", F.size(
    F.collect_set("artist").over(windowval)))
make_boxplot(df_valid, 'num_artists')<jupyter_output><empty_output><jupyter_text>Session<jupyter_code># number of sessions in a week
df_valid = df_valid.withColumn(
    'num_sessions', F.count('sessionId').over(windowval))
make_boxplot(df_valid, 'num_sessions')<jupyter_output><empty_output><jupyter_text>Status<jupyter_code>df_valid.groupby('status').count().show()
# flag the status 'redirect' and 'not found'
flag_redirect = F.udf(lambda x: 1 if x == 307 else 0, T.IntegerType())
flag_not_found = F.udf(lambda x: 1 if x == 404 else 0, T.IntegerType())
# create redirect and page_not_found columns
df_valid = df_valid.withColumn(
    "redirect", flag_redirect(df_valid['status']))
df_valid = df_valid.withColumn(
    "page_not_found", flag_not_found(df_valid['status']))
df_valid.groupBy('churn_user').sum('redirect').show()
df_valid.groupBy('churn_user').sum('page_not_found').show()
df_valid.filter(df_valid['status'] == '307').select(
    F.sum("churn_event")).show()
df_valid.filter(df_valid['status'] == '200').select(
    F.sum("churn_event")).show()<jupyter_output>+----------------+
|sum(churn_event)|
+----------------+
|              52|
+----------------+<jupyter_text>Page<jupyter_code>df_valid.groupby('page').count().show(22, False)
# create aggregate features from the column 'page'

pages = ['Thumbs Down', 'Thumbs Up', 'Home', 'Roll Advert',
         'Logout', 'Add to Playlist', 'Add Friend', 'Error']

for page in pages:
    flag_page = F.udf(lambda x: 1 if x == page else 0, T.IntegerType())
    df_valid = df_valid.withColumn('flag', flag_page('page'))
    df_valid = df_valid.withColumn(
        'num_' + page, F.sum('flag').over(windowval)).drop('flag')
make_boxplot(df_valid, 'num_Thumbs Down')
plt.savefig('num_Thumbs_Down')
make_boxplot(df_valid, 'num_Thumbs Up')
make_boxplot(df_valid, 'num_Home')
make_boxplot(df_valid, 'num_Roll Advert')
plt.savefig('num_Roll_Advert')
make_boxplot(df_valid, 'num_Logout')
make_boxplot(df_valid, 'num_Add to Playlist')
make_boxplot(df_valid, 'num_Add Friend')
plt.savefig('num_Add_Friend')
make_boxplot(df_valid, 'num_Error')
df_temp = df_valid.select(
    'churn_user',
    'itemInSession',
    'length',
    'days_since_reg',
    'hour',
    'day',
    'month',
    'num_songs',
    'num_artists',
    'num_sessions',
    'num_Thumbs Down',
    'num_Thumbs Up',
    'num_Home',
    'num_Roll Advert',
    'num_Logout',
    'num_Add to Playlist',
    'num_Add Friend',
    'num_Error'
).toPandas()
plt.figure(figsize=(15, 12))
sns.heatmap(df_temp.corr().abs(), annot=True, cmap='coolwarm');
#plt.ylim(12, 0)<jupyter_output><empty_output><jupyter_text>Highly correlated features will be dropped to avoid multicollinearity.<jupyter_code>df_temp_2 = df_valid.select(
    'churn_user',
    'itemInSession',
    'days_since_reg',
    'hour',
    'day',
    'month',
    'num_Thumbs Down',
    'num_Roll Advert',
    'num_Add Friend',
    'num_Error'
).toPandas()
plt.figure(figsize=(15, 12))
ax = sns.heatmap(df_temp_2.corr().abs(), annot=True, cmap='coolwarm');
plt.xticks(fontsize=16)
plt.yticks(fontsize=16)
plt.tight_layout()

plt.savefig('heatmap')<jupyter_output><empty_output><jupyter_text>Feature EngineeringData preprocessing and preparing machine learning pipelines for logistic regression, random forest and gradient boosted trees models. Stages for the pipelines include indexing and one-hot encoding categorical columns, transforming columns into a single vector column, and scaling the values of the features into the range from 0 to 1.<jupyter_code>def clean_data(df):
    """Return valid data by dropping missing values in registration."""
    df_clean = df.dropna(subset=['registration']).na.fill(
        0, subset=['length'])

    return df_clean


def is_churn(df):
    """Return dataframe with labels for churn users."""

    flag_churn_event = F.udf(
        lambda x: 1 if x == 'Cancellation Confirmation' else 0, T.IntegerType())

    df_new = df.withColumn(
        'churn_event', flag_churn_event(F.col('page')))

    df_new = df_new.withColumn('label', F.max(
        'churn_event').over(Window.partitionBy('userId')))

    return df_new


def registration_days(df):
    """Return the number of days since registration."""

    df_new = df.withColumn('registration date', (F.col(
        'registration')/1000).cast(T.TimestampType()))

    df_new = df_new.withColumn(
        'date', (F.col('ts')/1000).cast(T.TimestampType()))

    df_new = df_new.withColumn('days_since_registration', (F.datediff(
        'date', 'registration date'))).drop('date', 'registration date')

    return df_new


def timestamp(df):
    """Return dataframe with hour, day, and month."""

    get_hour = F.udf(lambda x: datetime.datetime.fromtimestamp(
        x / 1000.0).hour, T.IntegerType())
    get_day = F.udf(lambda x: datetime.datetime.fromtimestamp(
        x / 1000.0).day, T.IntegerType())
    get_month = F.udf(lambda x: datetime.datetime.fromtimestamp(
        x / 1000.0).month, T.IntegerType())

    df_new = df.withColumn('hour', get_hour(F.col('ts'))).withColumn(
        'day', get_day(F.col('ts'))).withColumn('month', get_month(F.col('ts')))

    return df_new

# time window to find aggregate values per day


def days(x):
    """Return time in seconds for the input time in days."""
    time_in_seconds = x * 86400
    return time_in_seconds


windowval = (
    Window.partitionBy("userId")
    .orderBy(F.col("ts")/1000)
    .rangeBetween(-days(7), 0)
)


# create aggregate features from the column 'page'
pages = ['Thumbs Down', 'Roll Advert', 'Add Friend', 'Error']


def aggregate_features(df):
    """Return selected aggregate features from page."""
    df_new = df

    for page in pages:

        flag_page = F.udf(lambda x: 1 if x ==
                          page else 0, T.IntegerType())
        df_new = df_new.withColumn('flag', flag_page('page'))
        df_new = df_new.withColumn(
            'num_' + page, F.sum('flag').over(windowval)).drop('flag')

    return df_new


def retrive_state(df):
    """Return dataframe with states retrived from locations."""
    df_new = df
    get_state = F.udf(lambda x: x[-2:])
    df_new = df_new.withColumn("state", get_state(
        df_new['location'])).drop('location')

    return df_new


ignore_list = ['artist',
               'auth',
               'firstName',
               'length',
               'lastName',
               'method',
               'page',
               'registration',
               'sessionId',
               'song',
               'ts',
               'userAgent',
               'userId',
               'churn_event',
               ]


def ignore(df, ignore_list):
    """Remove unnecessary feature columns."""
    df_new = df.drop(*ignore_list)

    return df_new


def preprocess_data(df):
    """
      1. Remove invalid data.
      2. Add churn event and churn user labels.
      3. Add number of days since registration.
      4. Add hour, day, and month.
      5. Add aggregate features.
      6. Get state from location.
      7. Remove unnecessary feature columns.
    """

    df_clean = clean_data(df)
    df_churn = is_churn(df_clean)
    df_reg_days = registration_days(df_churn)
    df_timestamp = timestamp(df_reg_days)
    df_aggregate = aggregate_features(df_timestamp)
    df_state = retrive_state(df_aggregate)
    df_processed = ignore(df_state, ignore_list)

    return df_processed


df_processed = preprocess_data(df).cache()
df.unpersist()


# Prepare stages for ML Pipeline
stages = []

# Index and one-hot encode categorical columns
cat_columns = ['gender', 'level', 'status', 'state', 'month']

for column in cat_columns:
    indexer = StringIndexer(inputCol=column, outputCol=column+'_index')
    encoder = OneHotEncoderEstimator(
        inputCols=[indexer.getOutputCol()], outputCols=[column + '_vect'])

    stages += [indexer, encoder]

num_columns = [
    'itemInSession',
    'days_since_registration',
    'hour',
    'day',
    'num_Thumbs Down',
    'num_Roll Advert',
    'num_Add Friend',
    'num_Error'
]

# Transform columns into a single vector column.
vect_assembler_inputs = [c + '_vect' for c in cat_columns] + num_columns
assembler = VectorAssembler(
    inputCols=vect_assembler_inputs, outputCol="features")

stages += [assembler]

# Rescale  each feature into the range from 0 to 1.
scaler = MinMaxScaler(inputCol="features", outputCol="scaled_features")

stages += [scaler]

# Instantiate classifiers
lr = LogisticRegression(labelCol='label', featuresCol='scaled_features')

rfc = RandomForestClassifier(
    labelCol='label', featuresCol='scaled_features')

gbt = GBTClassifier(labelCol='label', featuresCol='scaled_features')

# Set stages for classifiers
stages_lr = stages + [lr]
stages_rfc = stages + [rfc]
stages_gbt = stages + [gbt]
print('df_processed shape: {}, {}'.format(df_processed.count(), len(df_processed.columns)))
df_processed.columns<jupyter_output><empty_output><jupyter_text>ModelingSplit the full dataset into train and test datasets, set stages for pipelines and train models.<jupyter_code># Split the full dataset into train and test sets
train, test = df_processed.randomSplit([0.8, 0.2], seed=42)
# set pipelines
pipeline_lr = Pipeline().setStages(stages_lr)
pipeline_rfc = Pipeline().setStages(stages_rfc)
pipeline_gbt = Pipeline().setStages(stages_gbt)
# train models
model_lr = pipeline_lr.fit(train)
model_rfc = pipeline_rfc.fit(train)
model_gbt = pipeline_gbt.fit(train)<jupyter_output><empty_output><jupyter_text>Model Evaluation<jupyter_code># Instantiate evaluators
f1_eval = MulticlassClassificationEvaluator(
    labelCol='label', metricName='f1')  # F1 score evaluator

AUC_eval = BinaryClassificationEvaluator(
    labelCol='label', metricName="areaUnderROC")  # AUC evaluator
def evaluate_model(model):
    """ Return f1 and AUC scores of the model evaluated with test data."""
    preds = model.transform(test)
    f1 = f1_eval.evaluate(preds)
    AUC = AUC_eval.evaluate(preds)
    
    return f1, AUC
lr_f1, lr_AUC = evaluate_model(model_lr)
print('Logistic Regression')
print('F1 =', lr_f1)
print('AUC =', lr_AUC)
rfc_f1, rfc_AUC = evaluate_model(model_rfc)
print('Random Forest Classifier')
print('F1 =', rfc_f1)
print('AUC =', rfc_AUC)
gbt_f1, gbt_AUC = evaluate_model(model_gbt)
print('Gradient-boosted Tree Classifier')
print('F1 =', gbt_f1)
print('AUC =', gbt_AUC)
# create a dataframe for metrics
d = {'Classifier': ['LogisticRegression', 'RandomForestClassifier','GBTClassifier'],
     'F1': [lr_f1, rfc_f1, gbt_f1], 
     'AUC': [lr_AUC, rfc_AUC, gbt_AUC]}

df_metrics = pd.DataFrame(data=d)
df_metrics<jupyter_output><empty_output><jupyter_text>Model SelectionHyperparameter tuning for GBTClassifier using three-fold cross validation.<jupyter_code># Create param_grid for Cross Validation
param_grid = (ParamGridBuilder()
              .addGrid(gbt.stepSize, [0.1, 0.5])
              .addGrid(gbt.maxIter, [20, 32])
              .build())
pipeline = Pipeline().setStages(stages)

# Fit the pipeline with training data and transform train and test datasets
train_prep = pipeline.fit(train).transform(train).cache()
test_prep =pipeline.fit(train).transform(test).cache()
cv = CrossValidator(
    estimator=gbt,
    estimatorParamMaps=param_grid,
    evaluator=f1_eval,
    numFolds=2)

cv_model = cv.fit(train_prep)
# extract best model's parameters
cv_model.bestModel.extractParamMap()
cv_preds = cv_model.transform(test_prep)
cv_f1 = f1_eval.evaluate(cv_preds)
print('best_model F1 score:', cv_f1)<jupyter_output>best_model F1 score: 0.9752454479324296<jupyter_text>Classification Report<jupyter_code>cv_preds_pd = cv_preds.toPandas()
labels = cv_preds_pd['label']
predictions = cv_preds_pd['prediction']
probabilities = cv_preds_pd['probability'].apply(lambda x: x[1]).values
print(classification_report(labels, predictions))
confusion_matrix(labels, predictions)
# Create ROC curve
fpr, tpr, _ = roc_curve(labels, probabilities)

plt.plot(fpr, tpr)
plt.xlabel('False positive rate')
plt.ylabel('True positive rate')
plt.title('ROC curve');<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.17747146241654102, 'fraction_numerical': 0.018845573982339006, 'mean_word_length': 4.214111922141119, 'pattern_counts': {'":': 0, '<': 87, '<?xml version=': 0, '>': 87, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '23', 'dir': 'jupyter-structured-clean-dedup', 'id': '509661', 'license': 'MIT', 'n_tokens_mistral': 9982, 'n_tokens_neox': 9444, 'n_words': 1889, 'path': 'Sparkify.ipynb', 'repo_name': 'alwz1/Sparkify', 'size': '27858'}