<jupyter_start><jupyter_text>Spark MLLib - Random Forest **Description** - It is one of the most popular algorithms.- It is an Ensemble Method algorithm.- The Random Forest model builds several models and each one is used to predict results individually. In the end, Random Forest takes a vote to choose the best model. **Pros:** It usually offers good accuracy, is efficient with many predictor variables, works very well in parallel and is excellent with missing values. **Cons:** It is slower and Bias can occur frequently. **Application:** Scientific research, medical diagnosis.  Classifying a Bank's Credit Risk<jupyter_code>import math

from pyspark.ml.classification import RandomForestClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator
from pyspark.ml.feature import PCA
from pyspark.ml.feature import StringIndexer
from pyspark.ml.linalg import Vectors
from pyspark.sql import Row
spSession = SparkSession.builder.master('local').appName('BankCreditRisk').getOrCreate()
rddBank01 = sc.textFile('aux/datasets/bank.csv')<jupyter_output><empty_output><jupyter_text>**We can cache the RDD to optimize performance.**<jupyter_code>rddBank01.cache()
rddBank01.count()
rddBank01.take(5)
header = rddBank01.first()
rddBank02 = rddBank01.filter(lambda row: row != header)
rddBank02.count()<jupyter_output><empty_output><jupyter_text>Data Cleaning<jupyter_code>def dataCleaning(strRow):
    listAttr = strRow.replace('\"', '').split(';')
    
    age       = float(listAttr[0])
    balance   = float(listAttr[5])
    single    = 1.0 if listAttr[2]  == "single"    else 0.0
    married   = 1.0 if listAttr[2]  == "married"   else 0.0
    divorced  = 1.0 if listAttr[2]  == "divorced"  else 0.0
    primary   = 1.0 if listAttr[3]  == "primary"   else 0.0
    secondary = 1.0 if listAttr[3]  == "secondary" else 0.0
    tertiary  = 1.0 if listAttr[3]  == "tertiary"  else 0.0
    default   = 0.0 if listAttr[4]  == "no"        else 1.0
    loan      = 0.0 if listAttr[7]  == "no"        else 1.0
    outcome   = 0.0 if listAttr[16] == "no"        else 1.0
    
    row = Row(
        AGE = age,
        BALANCE  = balance,
        SINGLE = single,
        MARRIED  = married,
        DIVORCED = divorced,
        PRIMARY = primary,
        SECONDARY = secondary,
        TERTIARY = tertiary,
        DEFAULT = default,
        LOAN = loan,
        OUTCOME = outcome
    )
    
    return row
rddBank03 = rddBank02.map(dataCleaning)
rddBank03.collect()[:5]<jupyter_output><empty_output><jupyter_text>Exploratory Data Analysis **Converting the RDD to a DataFrame**<jupyter_code>dfBank = spSession.createDataFrame(rddBank03)
dfBank.describe(['AGE', 'BALANCE', 'SINGLE', 'MARRIED', 'DIVORCED']).show()
dfBank.describe(['DEFAULT', 'LOAN', 'OUTCOME']).show()
dfBank.describe(['DEFAULT', 'LOAN', 'OUTCOME']).show()
for column in dfBank.columns:
    if not(isinstance(dfBank.select(column).take(1)[0][0], str)):
        print(f"OUTCOME correlation with {column}: {dfBank.stat.corr('OUTCOME', column)}")<jupyter_output>OUTCOME correlation with AGE: -0.18232104327365253
OUTCOME correlation with BALANCE: 0.03657486611997681
OUTCOME correlation with SINGLE: 0.46323284934360515
OUTCOME correlation with MARRIED: -0.37532412991335623
OUTCOME correlation with DIVORCED: -0.07812659940926987
OUTCOME correlation with PRIMARY: -0.12561548832677982
OUTCOME correlation with SECONDARY: 0.026392774894072973
OUTCOME correlation with TERTIARY: 0.08494840766635618
OUTCOME correlation with DEFAULT: -0.04536965206737378
OUTCOME correlation with LOAN: -0.030420586112717318
OUTCOME correlation with OUTCOME: 1.0<jupyter_text>Data Pre-Processing **Creating a LabeledPoint (target, Vector[features])**It removes not relevant columns to the model (or with low correlation)<jupyter_code>def setLabeledPoint(row):
    labeledPoint = (
        row['OUTCOME'], 
        Vectors.dense([
            row['AGE'],
            row['BALANCE'],
            row['SINGLE'],
            row['MARRIED'],
            row['DIVORCED'],
            row['PRIMARY'],
            row['SECONDARY'],
            row['TERTIARY'],
            row['DEFAULT'],
            row['LOAN']
        ])
    )
    
    return labeledPoint
rddBank04 = dfBank.rdd.map(setLabeledPoint)
rddBank04.take(5)
dfBank = spSession.createDataFrame(rddBank04, ['label', 'features'])
dfBank.select('label', 'features').show(5)<jupyter_output>+-----+--------------------+
|label|            features|
+-----+--------------------+
|  0.0|[30.0,1787.0,0.0,...|
|  1.0|[33.0,4789.0,0.0,...|
|  1.0|[35.0,1350.0,1.0,...|
|  1.0|[30.0,1476.0,0.0,...|
|  0.0|[59.0,0.0,0.0,1.0...|
+-----+--------------------+
only showing top 5 rows<jupyter_text>**Applying PCA**<jupyter_code>pcaBank = PCA(k = 3, inputCol = 'features', outputCol = 'pca_features')
pcaModel = pcaBank.fit(dfBank)
pcaResult = pcaModel.transform(dfBank).select('label', 'pca_features')
pcaResult.show(5, truncate = False)<jupyter_output>+-----+-------------------------------------------------------------+
|label|pca_features                                                 |
+-----+-------------------------------------------------------------+
|0.0  |[-1787.018897197381,28.86209683775529,0.06459982604876331]   |
|1.0  |[-4789.020177138492,29.922562636341947,0.9830243513096395]   |
|1.0  |[-1350.022213163262,34.10110809796688,-0.8951427168301695]   |
|1.0  |[-1476.0189517184556,29.051333993596703,-0.39527238680219323]|
|0.0  |[-0.037889185366442445,58.9897182000177,0.7290792383661886]  |
+-----+-------------------------------------------------------------+
only showing top 5 rows<jupyter_text>**Creating a numeric index for the label target column**<jupyter_code>stringIndexer = StringIndexer(inputCol = 'label', outputCol = 'indexed')
stringIndexerModel = stringIndexer.fit(pcaResult)
dfBank = stringIndexerModel.transform(pcaResult)
dfBank.take(5)<jupyter_output><empty_output><jupyter_text>Machine Learning<jupyter_code>(dataTraining, dataTest) = dfBank.randomSplit([.7, .3])
dataTraining.count()
dataTest.count()
dataTraining.count() + dataTest.count() == dfBank.count()
randomForestClassifier = RandomForestClassifier(labelCol = 'indexed', featuresCol = 'pca_features')
model = randomForestClassifier.fit(dataTraining)
model
print(f'Features number: {str(model.numFeatures)}')
print(f'Classses number: {str(model.numClasses)}')
predictions = model.transform(dataTest)
predictions.select('label', 'indexed', 'pca_features', 'prediction').show(5)
evaluator = MulticlassClassificationEvaluator(
    predictionCol = 'prediction', 
    labelCol = 'indexed', 
    metricName = 'accuracy')
evaluator.evaluate(predictions)<jupyter_output><empty_output><jupyter_text>**Confusion Matrix - Summing Up Predictions**<jupyter_code>predictions.groupBy('indexed', 'prediction').count().show()<jupyter_output>+-------+----------+-----+
|indexed|prediction|count|
+-------+----------+-----+
|    1.0|       1.0|   18|
|    0.0|       1.0|   12|
|    1.0|       0.0|   39|
|    0.0|       0.0|   96|
+-------+----------+-----+
{'fraction_non_alphanumeric': 0.17918088737201365, 'fraction_numerical': 0.08845278725824801, 'mean_word_length': 3.2675970873786406, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '9', 'dir': 'jupyter-structured-clean-dedup', 'id': '578479', 'license': 'MIT', 'n_tokens_mistral': 2751, 'n_tokens_neox': 2396, 'n_words': 482, 'path': 'modules/09-apache-spark-machine-learning/03-spark-mllib-random-forest.ipynb', 'repo_name': 'cfascina/rtaps', 'size': '7032'}