<jupyter_start><jupyter_text>Sparkify Project - Cluster<jupyter_code># install pandas, matplotlib, and scikit-learn
sc.install_pypi_package("pandas==1.0.1") 
sc.install_pypi_package("matplotlib", "https://pypi.org/simple") 
sc.install_pypi_package("scikit-learn")
# 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
# Create spark session
spark = SparkSession \
    .builder \
    .appName("Sparkify") \
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>Load and Clean Dataset<jupyter_code># Read in full sparkify dataset
event_data = "s3n://udacity-dsnd/sparkify/sparkify_event_data.json"
df = spark.read.json(event_data)
df.head()
# Check Schema
df.printSchema()
# Check the number of records
df.count()
# 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
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><empty_output><jupyter_text>`artist`, `length`, and `song` will be null when a user is visiting pages other than `NextSong`.<jupyter_code># Check for the page when the 'artist' is not null
df_valid.filter(F.col('artist').isNotNull()).select(
    'page').dropDuplicates().show()
# Number of users
num_users = df_valid.select("userId").dropDuplicates().count()
print(num_users)<jupyter_output><empty_output><jupyter_text>Feature EngineeringClean data, define user churn and create aggregate features. Prepare stages such as indexing and one-hot encoding categorical features, and transforming feature columns into a single column for ML pipeline.<jupyter_code>def clean_data(df):
    """Return valid data by dropping missing values in registration and imputing length."""
    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 avg_itemInSession(df):
    """ Return datafram with average itemInSession. """

    df_new = df.withColumn('avg_itemInSession', F.avg(
        'itemInSession').over(Window.partitionBy('userId')))

    return df_new


def registration_days(df):
    """Return the maximum 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.max(F.datediff(
        'date', 'registration date')).over(Window.partitionBy('userId'))).drop('registration date', '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', F.avg(get_hour(F.col('ts'))).over(Window.partitionBy('userId')))\
        .withColumn('day', F.avg(get_day(F.col('ts'))).over(Window.partitionBy('userId')))\
        .withColumn('month', F.avg(get_month(F.col('ts'))).over(Window.partitionBy('userId')))

    return df_new


# 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(Window.partitionBy('userId'))).drop('flag')

    return df_new


def level_phase(df):
    """ Return dataframe with phases of users' subscription level."""

    df_new = df
    flag_downgrade_event = F.udf(
        lambda x: 1 if x == "Submit Downgrade" else 0, T.IntegerType())
    df_new = df_new.withColumn(
        "downgraded", flag_downgrade_event("page"))

    windowval = Window.partitionBy("userId").orderBy(
        F.desc("ts")).rangeBetween(Window.unboundedPreceding, 0)
    df_new = df_new.withColumn(
        "phase", F.sum("downgraded").over(windowval))

    return df_new


def retrive_state(df):
    """Return dataframe with uesrs' last 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']))

    return df_new


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


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 average items in session.
      4. Add number of days since registration.
      5. Add hour, day, and month.
      6. Add phase of scription level.
      7. Add aggregate features.
      8. Get state from location.
      9. Remove unnecessary feature columns.
    """

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

    return df_processed


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


# Prepare stages for ML Pipeline
stages = []

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

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 = [
    'avg_itemInSession',
    'days_since_registration',
    'hour',
    'day',
    'month',
    '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',
                    maxDepth=7, maxIter=50, stepSize=0.8, subsamplingRate=0.8)

# Set stages for classifiers
stages_lr = stages + [lr]
stages_rfc = stages + [rfc]
stages_gbt = stages + [gbt]
df_processed.groupBy('label').count().show()<jupyter_output><empty_output><jupyter_text>ModelingThree different models are trained and evaluated with full dataset which is split into 80% training data and 20% testing data. ML pipelines with stages as described in the feature engineering section are fitted and transformed. Three different models are trained and evaluated with F1 scores and AUC scores. The classifiers for the models are Logistic Regression, Random Forest Classifier, and Gradient Boosted Tree Classifier.<jupyter_code># Split the full dataset into train and test sets.
train, test = df_processed.randomSplit([0.8, 0.2], seed=42)
train.cache()
test.cache()
df_processed.unpersist()
# Set the stages for ML Pipeline
pipeline_lr = Pipeline().setStages(stages_lr)
pipeline_rfc = Pipeline().setStages(stages_rfc)
pipeline_gbt = Pipeline().setStages(stages_gbt)
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 Evaluations<jupyter_code># Instantiate evaluators

# F1 score evaluator
f1_eval = MulticlassClassificationEvaluator(
    labelCol='label', metricName='f1')

# areaUnderROC AUC evaluator
AUC_eval = BinaryClassificationEvaluator(
    labelCol='label', metricName="areaUnderROC")
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<jupyter_output><empty_output><jupyter_text>Logistic Regression<jupyter_code>lr_f1, lr_AUC = evaluate_model(model_lr)
print('Logistic Regression')
print('F1 =', lr_f1)
print('AUC =', lr_AUC)<jupyter_output><empty_output><jupyter_text>Random forest classifier<jupyter_code>rfc_f1, rfc_AUC = evaluate_model(model_rfc)
print('Random Forest Classifier')
print('F1 =', rfc_f1)
print('AUC =', rfc_AUC)<jupyter_output><empty_output><jupyter_text>Gradient-boosted tree classifier<jupyter_code>gbt_f1, gbt_AUC = evaluate_model(model_gbt)
print('Gradient-boosted Tree Classifier')
print('F1 =', gbt_f1)
print('AUC =', gbt_AUC)
preds = model_gbt.transform(test).select(
    'label', 'probability', 'prediction').toPandas()
labels = preds['label']
predictions = preds['prediction']
probabilities = preds['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.clf()

plt.plot(fpr, tpr)
plt.xlabel('False positive rate')
plt.ylabel('True positive rate')
plt.title('ROC curve');

%matplot plt<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11919887065466737, 'fraction_numerical': 0.007234868537144874, 'mean_word_length': 4.277001862197393, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 37, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '537096', 'license': 'MIT', 'n_tokens_mistral': 3762, 'n_tokens_neox': 3612, 'n_words': 871, 'path': 'Sparkify_aws.ipynb', 'repo_name': 'alwz1/Sparkify', 'size': '11334'}