<jupyter_start><jupyter_text>Sparkify-model build Dependencies: Sparkify-analysis and preprocessing has been run. This notebook contains the development of the Sparkify customer churn model.<jupyter_code>#Import libraries
from pyspark.sql import SparkSession
from pyspark.sql.functions import avg, col, asc, desc, lit, min, max, udf, isnull,when, row_number, floor,ceil, sum ,count, countDistinct
from pyspark.sql.types import IntegerType,DoubleType
from pyspark.sql import Window

from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression, NaiveBayes, GBTClassifier, RandomForestClassifier, LinearSVC
from pyspark.ml.evaluation import MulticlassClassificationEvaluator, BinaryClassificationEvaluator
from pyspark.ml.feature import MinMaxScaler, VectorAssembler
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder, CrossValidatorModel

import datetime
import matplotlib
import matplotlib.pyplot as plt
import pandas as pd
import seaborn as sns
import os
from time import time

import gc
#Check versions of packages
from platform import python_version

print(python_version())

print(pd.__version__)

print(matplotlib.__version__)
#Create spark session
spark = SparkSession \
    .builder \
    .appName("Sparkify") \
    .getOrCreate()

spark.sparkContext.getConf().getAll()
spark<jupyter_output><empty_output><jupyter_text>Load datasetLoad the dataset created in Sparkify-analysis and preprocessing `dfFinal.csv`.<jupyter_code>#Find current direct
current_path = os.getcwd()
current_path
dfFinal=pd.read_csv(str(current_path)+'/dfFinal.csv')
dfFinal
dfFinal = spark.createDataFrame(dfFinal)
dfFinal.printSchema()

#Ensure features are all of the right type for machine learning
featuresList = dfFinal.columns[:]
for feature in featuresList:
    dfFinal = dfFinal.withColumn(feature, dfFinal[feature].cast("float"))

#Drop unnecessary columns from features
dfMLInput = dfFinal.drop("userId")
feature_cols = dfMLInput.drop("userId", "churn").columns

dfMLInput<jupyter_output><empty_output><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># extract prediction columns (all except 'churned'), and split data into train and test sets
#pred_cols = users.columns
#pred_cols.remove('churned')
#train, test = users.randomSplit([0.75, 0.25],seed=123)



# Split data into train and test

traindf, testdf = dfMLInput.randomSplit([0.7, 0.3], seed = 42)
print('Shape of train and test sets:')
print((traindf.count(), len(traindf.columns)))
print((testdf.count(), len(testdf.columns)))<jupyter_output><empty_output><jupyter_text>Model selection for classicfication problem Have decided to train three models on the dataset:* Naive Bayes* Logistic regression * Naive Bayes* XG Boost* SVM  Training and predicting pipeline for preliminary models<jupyter_code>def train_predict(learner, train, test): 
    '''
    inputs:
       - learner: the learning algorithm to be trained and predicted on
       - train: features training set
       - test: features testing set
    '''
    #Create assembler and scaler for our pipeline
    #Transform features into features vector column
    assembler = VectorAssembler(inputCols=feature_cols,outputCol="featuresVector")

    #Scale the features    
    minmaxscaler = MinMaxScaler(inputCol="featuresVector", outputCol="scaledFeatures")

    #Performance metrics 
    evaluatorAcc = MulticlassClassificationEvaluator(labelCol='churn', predictionCol='prediction', metricName='accuracy')
    evaluatorF1 = MulticlassClassificationEvaluator(labelCol='churn', predictionCol='prediction', metricName='f1')
    evaluatorAUPR = BinaryClassificationEvaluator(labelCol='churn', metricName='areaUnderPR')
    
    #Pipeline for classifier using default parameters
    pipeline = Pipeline(stages=[assembler, minmaxscaler, learner])
    #results = {}
    
    #Train model and get predictions
    start = time() # Get start time

    model = pipeline.fit(train)
    preds = model.transform(test)
    
    end = time() # Get end time
    trainTime = end - start
    print('Total training time for {}: {} seconds'.format(learner.__class__.__name__, trainTime))
  
    
    #Calculate, tabulate and print metrics
    Acc = evaluatorAcc.evaluate(preds)
    F1 = evaluatorF1.evaluate(preds)
    AUPR = evaluatorAUPR.evaluate(preds)
    
    metrics.append([learner.__class__.__name__, Acc, F1, AUPR, trainTime])
    
    print('Accuracy Score for {} on test set: {} '.format(learner.__class__.__name__, Acc))   
    print('F1 Score for {} on test set: {} '.format(learner.__class__.__name__, F1))   
    print('Area under P-R curve for {} on test set: {} '.format(learner.__class__.__name__, AUPR))   

    return model,metrics<jupyter_output><empty_output><jupyter_text>Train and evaluate initial models<jupyter_code>metrics = []

modelNB=NaiveBayes(featuresCol='scaledFeatures', labelCol='churn')
modelLR=LogisticRegression(featuresCol='scaledFeatures', labelCol='churn')
modelRF=RandomForestClassifier(featuresCol='scaledFeatures', labelCol='churn',seed=42)
modelGBT=GBTClassifier(featuresCol='scaledFeatures', labelCol='churn',seed=42)
modelSVM=LinearSVC(featuresCol='scaledFeatures', labelCol='churn')

modelNB, metrics = train_predict(modelNB, traindf, testdf)
modelLR, metrics = train_predict(modelLR, traindf, testdf)
modelRF, metrics = train_predict(modelRF, traindf, testdf)
modelGBT, metrics = train_predict(modelGBT, traindf, testdf)
modelSVM, metrics = train_predict(modelSVM, traindf, testdf)
#Tabulate metrics

dfMetrics=pd.DataFrame(metrics)

dfMetrics.columns = ['Model','Accuracy','F1 Score','AUC P-R' ,'Training time']
dfMetrics

#Tabulate metrics

dfMetrics=pd.DataFrame(metrics)

dfMetrics.columns = ['Model','Accuracy','F1 Score','AUC P-R' ,'Training time']
dfMetrics<jupyter_output><empty_output><jupyter_text>Hyperparameter tuning for final model and evaluation<jupyter_code>RandomForestClassifier().explainParams()
RF = RandomForestClassifier(featuresCol='scaledFeatures', labelCol='churn',seed = 42)


#Create assembler and scaler for our pipeline
#Transform features into features vector column
assembler = VectorAssembler(inputCols=feature_cols,outputCol="featuresVector")

#Scale the features    
minmaxscaler = MinMaxScaler(inputCol="featuresVector", outputCol="scaledFeatures")

#Performance metrics 
evaluatorAcc = MulticlassClassificationEvaluator(labelCol='churn', predictionCol='prediction', metricName='accuracy')
evaluatorF1 = MulticlassClassificationEvaluator(labelCol='churn', predictionCol='prediction', metricName='f1')
evaluatorAUPR = BinaryClassificationEvaluator(labelCol='churn', metricName='areaUnderPR')
    
#Pipeline for classifier using default parameters
pipeline = Pipeline(stages=[assembler, minmaxscaler, RF])


paramGrid = ParamGridBuilder().addGrid(RF.impurity,['entropy', 'gini']) \
                              .addGrid(RF.maxDepth,[3,4,5])\
                              .addGrid(RF.maxBins,[5,10,15])\
                              .build()

cv = CrossValidator(estimator=pipeline,
                    estimatorParamMaps = paramGrid, 
                    evaluator = evaluatorAUPR,
                    numFolds=3,seed = 97)


#Train model and get predictions
start = time() # Get start time

modelFinal = cv.fit(traindf)
preds = modelFinal.transform(testdf)
    
end = time() # Get end time
trainTime = end - start
print('Total training time for {}: {} seconds'.format(RF.__class__.__name__, trainTime))
      
# Calculate, tabulate and print metrics
Acc = evaluatorAcc.evaluate(preds)
F1 = evaluatorF1.evaluate(preds)
AUPR = evaluatorAUPR.evaluate(preds)
    
metrics.append([modelFinal.__class__.__name__, Acc, F1, AUPR, trainTime])
    
print('Accuracy Score for {} on test set: {} '.format(RF.__class__.__name__, Acc))   
print('F1 Score for {} on test set: {} '.format(RF.__class__.__name__, F1))   
print('Area under P-R curve for {} on test set: {} '.format(RF.__class__.__name__, AUPR))   



#Tabulate metrics

dfMetrics=pd.DataFrame(metrics)

dfMetrics.columns = ['Model','Accuracy','F1 Score','AUC P-R' ,'Training time']

dfMetrics.loc[[2],["Model"]]="Baseline Random Forest"
dfMetrics.loc[[5],["Model"]]="Optimised Random Forest"
dfMetrics.iloc[[2,5],:]


def featureImportance(model):
    '''
    Rank the feature importance of a tree-based model.
    Inputs:
    - model: the trained tree-based model 
    Outputs:
    - table of the top ten most important features.
    '''
    featureImportanceList = list(model.bestModel.stages[-1].featureImportances)
    featuresList = list(feature_cols)

    importanceDict = {"Feature":featuresList,"importanceScore":featureImportanceList}
    pdFeatureImportance = pd.DataFrame(importanceDict).sort_values(by = "importanceScore", ascending = False)

    return (pdFeatureImportance.head(10))


top10 = featureImportance(modelFinal)
top10<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1049148312900076, 'fraction_numerical': 0.006618205489855701, 'mean_word_length': 5.361628709454797, 'pattern_counts': {'":': 2, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, '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': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '630308', 'license': 'IBM-pibs', 'n_tokens_mistral': 2827, 'n_tokens_neox': 2735, 'n_words': 702, 'path': 'Sparkify-model build.ipynb', 'repo_name': 'bev-o-neill/DSND-p7-Sparkify', 'size': '9217'}