<jupyter_start><jupyter_text>PySpark MLlib Live Assignment- The following assignment is based on [Machine Learning with PySpark and MLlib](https://towardsdatascience.com/machine-learning-with-pyspark-and-mllib-solving-a-binary-classification-problem-96396065d2aa)  Data Loader<jupyter_code>!wget https://archive.ics.uci.edu/ml/machine-learning-databases/00222/bank.zip
# Unzipping the bank.csv
!unzip "bank.zip"<jupyter_output>Archive:  bank.zip
replace bank-full.csv? [y]es, [n]o, [A]ll, [N]one, [r]ename: A
  inflating: bank-full.csv           
  inflating: bank-names.txt          
  inflating: bank.csv<jupyter_text>Exploring The Data- We will use the direct marketing campaigns (phone calls) of a Portuguese banking institution. The classification goal is to predict whether the client will subscribe (Yes/No) to a term deposit. The dataset can be downloaded from [Kaggle](https://www.kaggle.com/rouseguy/bankbalanced/data).<jupyter_code>!apt-get install openjdk-8-jdk-headless -qq > /dev/null

!wget -q https://downloads.apache.org/spark/spark-3.0.2/spark-3.0.2-bin-hadoop2.7.tgz

!tar xf spark-3.0.2-bin-hadoop2.7.tgz
!pip install -q findspark
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.0.2-bin-hadoop2.7"
import findspark
findspark.init()
findspark.find()
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('ml-bank').getOrCreate()
df = spark.read.csv('bank-full.csv', header = True, inferSchema = True, sep=';')
df.printSchema()<jupyter_output>root
 |-- age: integer (nullable = true)
 |-- job: string (nullable = true)
 |-- marital: string (nullable = true)
 |-- education: string (nullable = true)
 |-- default: string (nullable = true)
 |-- balance: integer (nullable = true)
 |-- housing: string (nullable = true)
 |-- loan: string (nullable = true)
 |-- contact: string (nullable = true)
 |-- day: integer (nullable = true)
 |-- month: string (nullable = true)
 |-- duration: integer (nullable = true)
 |-- campaign: integer (nullable = true)
 |-- pdays: integer (nullable = true)
 |-- previous: integer (nullable = true)
 |-- poutcome: string (nullable = true)
 |-- y: string (nullable = true)<jupyter_text>Input variables: age, job, marital, education, default, balance, housing, loan, contact, day, month, duration, campaign, pdays, previous, poutcome.Output variable: depositHave a peek of the first five observations. Pandas data frame is prettier than Spark DataFrame.show().<jupyter_code>import pandas as pd

pd.DataFrame(df.take(5), columns=df.columns).transpose()<jupyter_output><empty_output><jupyter_text>Summary statistics for numeric variables<jupyter_code>numeric_features = [t[0] for t in df.dtypes if t[1] == 'int']
df.select(numeric_features).describe().toPandas().transpose()<jupyter_output><empty_output><jupyter_text>Correlations between independent variables.<jupyter_code>numeric_data = df.select(numeric_features).toPandas()
axs = pd.plotting.scatter_matrix(numeric_data, figsize=(8, 8));
n = len(numeric_data.columns)

for i in range(n):
    v = axs[i, 0]
    v.yaxis.label.set_rotation(0)
    v.yaxis.label.set_ha('right')
    v.set_yticks(())
    h = axs[n-1, i]
    h.xaxis.label.set_rotation(90)
    h.set_xticks(())<jupyter_output><empty_output><jupyter_text>It’s obvious that there aren’t highly correlated numeric variables. Therefore, we will keep all of them for the model. However, day and month columns are not really useful, we will remove these two columns.<jupyter_code>df = df.select('age', 'job', 'marital', 'education', 'default', 'balance', 'housing', 'loan', 'contact', 'duration', 'campaign', 'pdays', 'previous', 'poutcome', 'y')
cols = df.columns
df.printSchema()<jupyter_output>root
 |-- age: integer (nullable = true)
 |-- job: string (nullable = true)
 |-- marital: string (nullable = true)
 |-- education: string (nullable = true)
 |-- default: string (nullable = true)
 |-- balance: integer (nullable = true)
 |-- housing: string (nullable = true)
 |-- loan: string (nullable = true)
 |-- contact: string (nullable = true)
 |-- duration: integer (nullable = true)
 |-- campaign: integer (nullable = true)
 |-- pdays: integer (nullable = true)
 |-- previous: integer (nullable = true)
 |-- poutcome: string (nullable = true)
 |-- y: string (nullable = true)<jupyter_text>Preparing Data for Machine Learning The process includes Category Indexing, One-Hot Encoding and VectorAssembler — a feature transformer that merges multiple columns into a vector column.<jupyter_code>from pyspark.ml.feature import OneHotEncoder , StringIndexer, VectorAssembler

categoricalColumns = ['job', 'marital', 'education', 'default', 'housing', 'loan', 'contact', 'poutcome']
stages = []

for categoricalCol in categoricalColumns:
    stringIndexer = StringIndexer(inputCol = categoricalCol, outputCol = categoricalCol + 'Index')
    encoder = OneHotEncoder(inputCols=[stringIndexer.getOutputCol()], outputCols=[categoricalCol + "classVec"])
    stages += [stringIndexer, encoder]

label_stringIdx = StringIndexer(inputCol = 'y', outputCol = 'label')
stages += [label_stringIdx]
numericCols = ['age', 'balance', 'duration', 'campaign', 'pdays', 'previous']
assemblerInputs = [c + "classVec" for c in categoricalColumns] + numericCols
### START CODE ###
# Use VectorAssembler and pass in assemblerInputs as inputCols and name the outputCol "features"
assembler = VectorAssembler(inputCols = assemblerInputs, outputCol = "features")
### END CODE ###
stages += [assembler]<jupyter_output><empty_output><jupyter_text>The above code is taken from [databricks’ official site](https://docs.databricks.com/applications/machine-learning/train-model/mllib/index.htmlbinary-classification-example) and it indexes each categorical column using the StringIndexer, then converts the indexed categories into one-hot encoded variables. The resulting output has the binary vectors appended to the end of each row. We use the StringIndexer again to encode our labels to label indices. Next, we use the VectorAssembler to combine all the feature columns into a single vector column.  Pipeline We use Pipeline to chain multiple Transformers and Estimators together to specify our machine learning workflow. A Pipeline’s stages are specified as an ordered array.<jupyter_code>from pyspark.ml import Pipeline
pipeline = Pipeline(stages = stages)
### START CODE ###
# Fit the pipeline on df
pipelineModel = pipeline.fit(df)
# Transform pipelineModel on df
df = df = pipelineModel.transform(df)
### END CODE ###
selectedCols = ['label', 'features'] + cols
df = df.select(selectedCols)
df.printSchema()
pd.DataFrame(df.take(5), columns=df.columns).transpose()
### START CODE ###
# Do a randomSplit with 0.7 and 0.3 with a seed 42
train, test = train, test = df.randomSplit([0.7, 0.3], seed = 42)
### END CODE ###
print("Training Dataset Count: " + str(train.count()))
print("Test Dataset Count: " + str(test.count()))<jupyter_output>Training Dataset Count: 31724
Test Dataset Count: 13487<jupyter_text>Logistic Regression Model- You can build a RandomForestClassifier with : from pyspark.ml.classification import RandomForestClassifier- You can build a Gradient-Boosted Tree Classifier with : from pyspark.ml.classification import GBTClassifier<jupyter_code>from pyspark.ml.classification import LogisticRegression

### START CODE ###
# Fit a LogisticRegression with featuresCol as 'features', labelCol as 'label' and a maxIter of 10
lr = LogisticRegression(featuresCol='features', labelCol='label', maxIter=10)
### END CODE ###
lrModel = lr.fit(train)<jupyter_output><empty_output><jupyter_text>We can obtain the coefficients by using LogisticRegressionModel’s attributes.<jupyter_code>import matplotlib.pyplot as plt
import numpy as np
beta = np.sort(lrModel.coefficients)
plt.plot(beta)
plt.ylabel('Beta Coefficients')
plt.show()<jupyter_output><empty_output><jupyter_text>Evaluate our Logistic Regression model.<jupyter_code>predictions = lrModel.transform(test)
predictions.select('age', 'job', 'label', 'rawPrediction', 'prediction', 'probability').show(10)
from pyspark.ml.evaluation import BinaryClassificationEvaluator

evaluator = BinaryClassificationEvaluator()
print('Test Area Under ROC', evaluator.evaluate(predictions))<jupyter_output>Test Area Under ROC 0.8886722185888365<jupyter_text>OPTIONAL: HyperParameter Tuning a Gradient-Boosted Tree Classifier<jupyter_code>from pyspark.ml.classification import GBTClassifier

gbt = GBTClassifier(maxIter=10)
gbtModel = gbt.fit(train)
predictions = gbtModel.transform(test)
predictions.select('age', 'job', 'label', 'rawPrediction', 'prediction', 'probability').show(10)
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
paramGrid = (ParamGridBuilder()
             .addGrid(gbt.maxDepth, [2, 4, 6])
             .addGrid(gbt.maxBins, [20, 60])
             .addGrid(gbt.maxIter, [10, 20])
             .build())
cv = CrossValidator(estimator=gbt, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=5)
# Run cross validations.  This can take about 6 minutes since it is training over 20 trees!
cvModel = cv.fit(train)
predictions = cvModel.transform(test)
evaluator.evaluate(predictions)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1272924579489962, 'fraction_numerical': 0.012045577862181226, 'mean_word_length': 5.467368421052631, 'pattern_counts': {'":': 0, '<': 44, '<?xml version=': 0, '>': 45, 'https://': 5, 'lorem ipsum': 0, 'www.': 1, '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': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '638484', 'license': 'MIT', 'n_tokens_mistral': 3023, 'n_tokens_neox': 2921, 'n_words': 804, 'path': 'PySpark_Live_Assignment_Exercise.ipynb', 'repo_name': 'VinayNooji/FourthBrain', 'size': '9215'}