<jupyter_start><jupyter_text>Models <NAME>, <NAME>, <NAME>, <NAME><jupyter_code>from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()
sc = spark.sparkContext
import pandas as pd
from pyspark.ml import Pipeline  
from pyspark.ml.feature import *  
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.classification import GBTClassifier, RandomForestClassifier, NaiveBayes,\
LinearSVC, MultilayerPerceptronClassifier
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import BinaryClassificationEvaluator
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
import matplotlib.pyplot as plt
%matplotlib inline
#We can use the sample.csv to speed things up while tuning models
df = spark.read.csv('train_data.csv', inferSchema=True, header=True)
#df = spark.read.csv('sample.csv', inferSchema=True, header=True)
df.show(2)
dfs = df.select([col for col in df.columns if col not in ['_c0']])
dfs.show(5)
dfs.printSchema()<jupyter_output>root
 |-- loan_status_binary: integer (nullable = true)
 |-- loan_amnt: double (nullable = true)
 |-- term: string (nullable = true)
 |-- int_rate: double (nullable = true)
 |-- home_ownership: string (nullable = true)
 |-- tot_cur_bal: double (nullable = true)
 |-- total_pymnt: double (nullable = true)
 |-- annual_inc: double (nullable = true)
 |-- addr_state: string (nullable = true)
 |-- fico_range_low: double (nullable = true)
 |-- last_pymnt_amnt: double (nullable = true)
 |-- debt_settlement_flag: string (nullable = true)
 |-- pub_rec_bankruptcies: double (nullable = true)
 |-- delinq_2yrs: double (nullable = true)
 |-- grade_feature: string (nullable = true)
 |-- emp_length_low: integer (nullable = true)<jupyter_text>Logistic Regression Note that for all models, we ran using 5 fold cross validation on a sample of the training set first to find the best parameters, then used those parameters on the full set. You will see the cross validation models commented out but they have been run.<jupyter_code>train, test = dfs.randomSplit([0.8, 0.2], seed=12345)
categories = ['home_ownership', 'term', 'debt_settlement_flag', 'grade_feature']
stages = []
for col in categories:
    stringIndexer = StringIndexer(inputCol=col, outputCol=col + "_Index")
    encoder = OneHotEncoder(inputCol=stringIndexer.getOutputCol(),
                            outputCol=col + "classVec")
    stages += [stringIndexer, encoder]
scaled_vectors = VectorAssembler(inputCols = ['loan_amnt',\
                                             'total_pymnt', 'annual_inc', \
                                             'fico_range_low', 'last_pymnt_amnt',\
                                             'int_rate','emp_length_low','tot_cur_bal',
                                             'delinq_2yrs', 'pub_rec_bankruptcies',\
                                             'chargeoff_within_12_mths'],
                                 outputCol='vector_features')
scaler = StandardScaler(inputCol='vector_features', outputCol='scaled_features')
labelIndexer = StringIndexer(inputCol='loan_status_binary', outputCol='label')
#AT added
assembler = VectorAssembler(inputCols=['scaled_features', 'home_ownershipclassVec', 'termclassVec', \
                                       'debt_settlement_flagclassVec','grade_featureclassVec' ], outputCol='features')
stages += [scaled_vectors, scaler, labelIndexer, assembler]
lr = LogisticRegression(maxIter=10, regParam=0.1)
stages += [lr]
stages
pipeline = Pipeline(stages=stages)
paramGrid = ParamGridBuilder() \
    .addGrid(lr.regParam, [0.1, 0.01, .001]) \
    .addGrid(lr.elasticNetParam, [0.0, 0.2, 0.8, 1.0])\
    .build()
crossval = CrossValidator(estimator=pipeline,
                          estimatorParamMaps=paramGrid,
                          evaluator=BinaryClassificationEvaluator(),
                          numFolds=5)
# Fit the pipeline
model = crossval.fit(train)
# Make a prediction
prediction = model.transform(test)
prediction.show(2)
prediction = prediction.withColumn("loan_status_binary", df["loan_status_binary"].cast('float'))
#This is necessary to compare predictions with actual values
#Calculates area under Precision-Recall curve
evaluator = BinaryClassificationEvaluator(labelCol='loan_status_binary', \
                                          metricName='areaUnderPR')
evaluator.evaluate(prediction) #calculates area under P-R curve
f1_evaluator = MulticlassClassificationEvaluator(predictionCol='prediction',\
                                                 labelCol='loan_status_binary',\
                                                 metricName='f1')
f1_evaluator.evaluate(prediction) #f1 score
trainingSummary = model.bestModel.stages[-1].summary
trainingSummary.accuracy
#There's probably a better way, but this shows the best parameters
bestPipeline = model.bestModel
bestLRModel = bestPipeline.stages[-1]
bestParams = bestLRModel.extractParamMap()
bestParams
#plotting precision-recall curve
pr = trainingSummary.pr.toPandas()

plt.plot(pr['recall'],pr['precision'])
plt.ylabel('Precision')
plt.xlabel('Recall')
plt.show()<jupyter_output><empty_output><jupyter_text>Confusion Matrix<jupyter_code>#from pyspark.mllib.evaluation import BinaryClassificationMetrics
from sklearn.metrics import confusion_matrix
#predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
y_true = prediction.select("loan_status_binary")
y_true = y_true.toPandas()

y_pred = prediction.select("prediction")
y_pred = y_pred.toPandas()

cnf_matrix = confusion_matrix(y_true, y_pred,labels=[0, 1])
cnf_matrix

import numpy as np
import itertools

def plot_confusion_matrix(cm, classes,
                          normalize=False,
                          title='Confusion matrix',
                          cmap=plt.cm.Blues):
    """
    This function prints and plots the confusion matrix.
    Normalization can be applied by setting `normalize=True`.
    """
    if normalize:
        cm = cm.astype('float') / cm.sum(axis=1)[:, np.newaxis]
        print("Normalized confusion matrix")
    else:
        print('Confusion matrix, without normalization')

    print(cm)

    plt.imshow(cm, interpolation='nearest', cmap=cmap)
    plt.title(title)
    plt.colorbar()
    tick_marks = np.arange(len(classes))
    plt.xticks(tick_marks, classes, rotation=45)
    plt.yticks(tick_marks, classes)

    fmt = '.2f' if normalize else 'd'
    thresh = cm.max() / 2.
    for i, j in itertools.product(range(cm.shape[0]), range(cm.shape[1])):
        plt.text(j, i, format(cm[i, j], fmt),
                 horizontalalignment="center",
                 color="white" if cm[i, j] > thresh else "black")

    plt.tight_layout()
    plt.ylabel('True label')
    plt.xlabel('Predicted label')
plt.figure()
plot_confusion_matrix(cnf_matrix, classes=[0, 1],
                      title='Confusion matrix')
plt.show()<jupyter_output>Confusion matrix, without normalization
[[395643   2712]
 [ 45557   8085]]<jupyter_text>Linear SVC<jupyter_code>stages[-1] = LinearSVC(regParam = 0.001)
pipeline = Pipeline(stages=stages)<jupyter_output><empty_output><jupyter_text>Note: In this and successive models, cross validation was run first to find the best parameters, then the full training set was run using the best parameter.<jupyter_code>#paramGrid = ParamGridBuilder() \
  #  .addGrid(lr.regParam, [0.1, 0.01, .001]) \
   # .build()
#crossval = CrossValidator(estimator=pipeline,
 #                         estimatorParamMaps=paramGrid,
  #                        evaluator=BinaryClassificationEvaluator(),
   #                       numFolds=5)
# Fit the pipeline
#model_svc = crossval.fit(train)
model_svc = pipeline.fit(train)
# Make a prediction
prediction = model_svc.transform(test)
print(f'Area under P-R Curve is {evaluator.evaluate(prediction)}')
print(f'F1 Score is {f1_evaluator.evaluate(prediction)}')
#predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
y_true = prediction.select("loan_status_binary")
y_true = y_true.toPandas()

y_pred = prediction.select("prediction")
y_pred = y_pred.toPandas()

cnf_matrix = confusion_matrix(y_true, y_pred,labels=[0, 1])
cnf_matrix
plt.figure()
plot_confusion_matrix(cnf_matrix, classes=[0, 1],
                      title='Confusion matrix')
plt.show()<jupyter_output>Confusion matrix, without normalization
[[398181    174]
 [ 47119   6523]]<jupyter_text>Gradient Boosted Trees<jupyter_code>categories = ['home_ownership', 'term', 'grade_feature', 'debt_settlement_flag']
stages = []


for col in categories:
    stringIndexer = StringIndexer(inputCol=col, outputCol=col + "_Index")
    encoder = OneHotEncoder(inputCol=stringIndexer.getOutputCol(),
                            outputCol=col + "classVec")
    stages += [stringIndexer, encoder]
vectors = VectorAssembler(inputCols = ['loan_amnt', 'home_ownershipclassVec', 'termclassVec',\
                                             'total_pymnt', 'annual_inc', \
                                             'fico_range_low', 'last_pymnt_amnt',\
                                             'int_rate','emp_length_low',\
                                       'delinq_2yrs', 'pub_rec_bankruptcies',\
                                      'grade_featureclassVec', 'debt_settlement_flagclassVec',\
                                      'chargeoff_within_12_mths'], 
                                 outputCol='features')
labelIndexer = StringIndexer(inputCol='loan_status_binary', outputCol='label')
gbt = GBTClassifier(maxDepth=12, maxBins=32, maxMemoryInMB=1024)
stages += [vectors, labelIndexer, gbt]
pipeline = Pipeline(stages=stages)
#paramGrid = ParamGridBuilder() \
    #.addGrid(gbt.maxDepth, [20]) \
    #.addGrid(gbt.maxBins, [50])\
    #.build()
#crossval = CrossValidator(estimator=pipeline,
                          #estimatorParamMaps=paramGrid,
                          #evaluator=BinaryClassificationEvaluator(),
                          #numFolds=5)
# Fit the pipeline
#model_gbt = crossval.fit(train)
model = pipeline.fit(train)
# Make a prediction
prediction = model.transform(test)
print(f'Area under P-R Curve is {evaluator.evaluate(prediction)}')
print(f'F1 Score is {f1_evaluator.evaluate(prediction)}')
#predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
y_true = prediction.select("loan_status_binary")
y_true = y_true.toPandas()

y_pred = prediction.select("prediction")
y_pred = y_pred.toPandas()

cnf_matrix = confusion_matrix(y_true, y_pred,labels=[0, 1])
cnf_matrix
plt.figure()
plot_confusion_matrix(cnf_matrix, classes=[0, 1],
                      title='Confusion matrix')
plt.show()<jupyter_output>Confusion matrix, without normalization
[[395289   3066]
 [ 29146  24496]]<jupyter_text>Random Forest<jupyter_code>rf = RandomForestClassifier(maxDepth=10, maxBins=25, numTrees=30, maxMemoryInMB=1024)
stages[-1] = rf
pipeline = Pipeline(stages=stages)
#paramGrid = ParamGridBuilder() \
    #.addGrid(rf.maxDepth, [5, 10]) \
    #.addGrid(rf.maxBins, [15, 20])\
    #.addGrid(rf.numTrees, [20, 25])\
    #.build()
#crossval = CrossValidator(estimator=pipeline,
 #                         estimatorParamMaps=paramGrid,
  #                        evaluator=BinaryClassificationEvaluator(),
   #                       numFolds=5)
# Fit the pipeline
#model_rf = crossval.fit(train)
model_rf = pipeline.fit(train)
# Make a prediction
prediction = model_rf.transform(test)
print(f'Area under P-R Curve is {evaluator.evaluate(prediction)}')
print(f'F1 Score is {f1_evaluator.evaluate(prediction)}')
#Find Best Parameters
bestPipeline = model_rf.bestModel
bestLRModel = bestPipeline.stages[-1]
bestParams = bestLRModel.extractParamMap()
bestParams
#As with GBT, it wants higher values for parameters to improve performance
#predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
y_true = prediction.select("loan_status_binary")
y_true = y_true.toPandas()

y_pred = prediction.select("prediction")
y_pred = y_pred.toPandas()

cnf_matrix = confusion_matrix(y_true, y_pred,labels=[0, 1])
cnf_matrix
plt.figure()
plot_confusion_matrix(cnf_matrix, classes=[0, 1],
                      title='Confusion matrix')
plt.show()<jupyter_output>Confusion matrix, without normalization
[[397900    455]
 [ 43640  10002]]<jupyter_text>Naive Bayes Model<jupyter_code>categories = ['home_ownership', 'term', 'debt_settlement_flag', 'grade_feature']
stages = []

labelIndexer = StringIndexer(inputCol='loan_status_binary', outputCol='label')

for col in categories:
    stringIndexer = StringIndexer(inputCol=col, outputCol=col + "_Index")
    encoder = OneHotEncoder(inputCol=stringIndexer.getOutputCol(),
                            outputCol=col + "classVec")
    stages += [stringIndexer, encoder]

vectors = VectorAssembler(inputCols = ['loan_amnt', 'home_ownershipclassVec', 'termclassVec',\
                                             'total_pymnt', 'annual_inc', \
                                             'fico_range_low', 'last_pymnt_amnt',\
                                             'int_rate','emp_length_low',\
                                      'grade_featureclassVec','debt_settlement_flagclassVec',\
                                      'delinq_2yrs', 'pub_rec_bankruptcies'], 
                                 outputCol='vector_features')
assembler = VectorAssembler(inputCols=['vector_features', 'home_ownershipclassVec', 'termclassVec', \
                                       'debt_settlement_flagclassVec','grade_featureclassVec' ], outputCol='features')
nb = NaiveBayes(labelCol="label", featuresCol="vector_features", smoothing=1.0)

stages += [vectors, labelIndexer, nb]

pipeline = Pipeline(stages=stages)
# Create ParamGrid for Cross Validation
#nbparamGrid = (ParamGridBuilder()
               #.addGrid(nb.smoothing, [0.0,0.2, 0.4, 0.6, 0.8, 1.0])
               #.build())

# Evaluate model
#nbevaluator = BinaryClassificationEvaluator(rawPredictionCol="rawPrediction")

# Create 5-fold CrossValidator
#nbcv = CrossValidator(estimator = pipeline,
 #                   estimatorParamMaps = nbparamGrid,
  #                  evaluator=BinaryClassificationEvaluator(),
                    #evaluator = nbevaluator,
   #                 numFolds = 5)
model_n_bayes = pipeline.fit(train)
prediction = model_n_bayes.transform(test)
bestPipeline = model_n_bayes.bestModel
bestNBModel = bestPipeline.stages[-1]
bestParams = bestNBModel.extractParamMap()
bestParams
print(f'Area under P-R Curve is {evaluator.evaluate(prediction)}')
print(f'F1 Score is {f1_evaluator.evaluate(prediction)}')
#predictionAndLabels = test.map(lambda lp: (float(model.predict(lp.features)), lp.label))
y_true = prediction.select("loan_status_binary")
y_true = y_true.toPandas()

y_pred = prediction.select("prediction")
y_pred = y_pred.toPandas()

cnf_matrix = confusion_matrix(y_true, y_pred,labels=[0, 1])
cnf_matrix
plt.figure()
plot_confusion_matrix(cnf_matrix, classes=[0, 1],
                      title='Confusion matrix')
plt.show()<jupyter_output>Confusion matrix, without normalization
[[184824 213531]
 [  2880  50762]]
{'fraction_non_alphanumeric': 0.11638641608968019, 'fraction_numerical': 0.016419386745796242, 'mean_word_length': 3.4370977179637214, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 32, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 38, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '621945', 'license': 'Apache-2.0', 'n_tokens_mistral': 4919, 'n_tokens_neox': 4677, 'n_words': 937, 'path': 'Project_Models.ipynb', 'repo_name': 'matthew-e-thomas/Spark_ML_DS5559', 'size': '15188'}