{"id":1519,"date":"2019-02-18T04:39:42","date_gmt":"2019-02-18T04:39:42","guid":{"rendered":"http:\/\/kusuaks7\/?p=1124"},"modified":"2023-08-09T14:16:29","modified_gmt":"2023-08-09T14:16:29","slug":"machine-learning-with-pyspark-and-mlib-solving-a-binary-classification-problem","status":"publish","type":"post","link":"https:\/\/www.experfy.com\/blog\/ai-ml\/machine-learning-with-pyspark-and-mlib-solving-a-binary-classification-problem\/","title":{"rendered":"Machine Learning with PySpark and MLlib\u200a\u2014\u200aSolving a Binary Classification Problem"},"content":{"rendered":"

Ready to learn Machine Learning? Browse courses<\/a>\u00a0like\u00a0Machine Learning Foundations: Supervised Learning<\/a> developed by industry thought leaders and Experfy in Harvard Innovation Lab.<\/em><\/strong><\/p>\n

<\/h2>\n

Photo Credit: Pixabay<\/p>\n

Apache Spark<\/a>, once a component of the\u00a0Hadoop<\/a>\u00a0ecosystem, is now becoming the big-data platform of choice for enterprises. It is a powerful open source engine that provides real-time stream processing, interactive processing, graph processing, in-memory processing as well as batch processing with very fast speed, ease of use and standard interface.<\/p>\n

In the industry, there is a big demand for a powerful engine that can do all of above. Sooner or later, your company or your clients will be using Spark to develop sophisticated models that would enable you to discover new opportunities or avoid risk. Spark is not hard to learn, if you already known Python and SQL, it is very easy to get started. Let\u2019s give it a try today!<\/p>\n

Exploring The\u00a0Data<\/strong><\/h3>\n

We will use the same data set when we\u00a0built a Logistic Regression in Python<\/a>, and it is related to 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\u00a0Kaggle<\/a>.<\/p>\n

from pyspark.sql import SparkSession<\/span>
\nspark = SparkSession.builder.appName(‘ml-bank’).getOrCreate()<\/span>
\ndf = spark.read.csv(‘bank.csv’, header = True, inferSchema = True)<\/span>
\ndf.printSchema()<\/span><\/span><\/p>\n

<\/p>\n

Figure 1<\/p>\n

Input variables: age, job, marital, education, default, balance, housing, loan, contact, day, month, duration, campaign, pdays, previous, poutcome.<\/p>\n

Output variable: deposit<\/p>\n

Have a peek of the first five observations. Pandas data frame is prettier than Spark DataFrame.show().<\/p>\n

import pandas as pd<\/span>
\npd.DataFrame(df.take(5), columns=df.columns).transpose()<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 2<\/p>\n

Our classes are perfect balanced.<\/p>\n

import pandas as pd<\/span>
\npd.DataFrame(df.take(5), columns=df.columns).transpose()<\/span>\u00a0<\/span><\/p>\n

<\/p>\n

Figure 3<\/p>\n

Summary statistics for numeric variables<\/strong><\/p>\n

numeric_features = [t[0] for t in df.dtypes if t[1] == ‘int’]<\/span>
\ndf.select(numeric_features).describe().toPandas().transpose()<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 4<\/p>\n

Correlations between independent variables<\/strong>.<\/p>\n

numeric_data = df.select(numeric_features).toPandas()<\/span><\/span><\/p>\n

axs = pd.scatter_matrix(numeric_data, figsize=(8, 8));<\/span><\/span><\/p>\n

n = len(numeric_data.columns)<\/span>
\nfor i in range(n):<\/span>
\n\u00a0\u00a0\u00a0 v = axs[i, 0]<\/span>
\n\u00a0\u00a0\u00a0 v.yaxis.label.set_rotation(0)<\/span>
\n\u00a0\u00a0\u00a0 v.yaxis.label.set_ha(‘right’)<\/span>
\n\u00a0\u00a0\u00a0 v.set_yticks(())<\/span>
\n\u00a0\u00a0\u00a0 h = axs[n-1, i]<\/span>
\n\u00a0\u00a0\u00a0 h.xaxis.label.set_rotation(90)<\/span>
\n\u00a0\u00a0\u00a0 h.set_xticks(())<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 5<\/p>\n

It\u2019s obvious that there aren\u2019t 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.<\/p>\n

df = df.select(‘age’, ‘job’, ‘marital’, ‘education’, ‘default’, <\/span><\/span><\/div>\n
‘balance’, ‘housing’, ‘loan’, ‘contact’, ‘duration’, ‘campaign’,<\/span><\/span><\/div>\n
‘pdays’, ‘previous’, ‘poutcome’, ‘deposit’)<\/span>
\ncols = df.columns<\/span>
\ndf.printSchema()<\/span><\/span><\/div>\n
<\/canvas><\/figure>\n

Figure 6<\/p>\n

Preparing Data for Machine\u00a0Learning<\/strong><\/h3>\n

The process includes Category Indexing, One-Hot Encoding and VectorAssembler \u2014 a feature transformer that merges multiple columns into a vector column.<\/p>\n

from pyspark.ml.feature import OneHotEncoderEstimator, <\/span>
\nStringIndexer, VectorAssembler<\/span><\/span><\/p>\n

categoricalColumns = [‘job’, ‘marital’, ‘education’, ‘default’, <\/span>
\n‘housing’, ‘loan’, ‘contact’, ‘poutcome’]<\/span>
\nstages = []<\/span><\/span><\/p>\n

for categoricalCol in categoricalColumns:<\/span>
\n\u00a0\u00a0\u00a0 stringIndexer = StringIndexer(inputCol = categoricalCol, <\/span>
\noutputCol = categoricalCol + ‘Index’)<\/span>
\n\u00a0\u00a0\u00a0 encoder = OneHotEncoderEstimator(inputCols=<\/span>
\n[stringIndexer.getOutputCol()], outputCols=[categoricalCol +<\/span>
\n\u00a0“classVec”])<\/span>
\n\u00a0\u00a0\u00a0 stages += [stringIndexer, encoder]<\/span><\/span><\/p>\n

label_stringIdx = StringIndexer(inputCol = ‘deposit’, outputCol = ‘label’)<\/span>
\nstages += [label_stringIdx]<\/span><\/span><\/p>\n

numericCols = [‘age’, ‘balance’, ‘duration’, ‘campaign’, ‘pdays’,<\/span>
\n\u00a0‘previous’]<\/span>
\nassemblerInputs = [c + “classVec” for c in categoricalColumns] +<\/span>
\n\u00a0numericCols<\/span>
\nassembler = VectorAssembler(inputCols=assemblerInputs, <\/span>
\noutputCol=”features”)<\/span>
\nstages += [assembler]<\/span><\/span><\/p>\n

The above code are taken from\u00a0databricks\u2019 official site<\/a>\u00a0and 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.<\/p>\n

Pipeline<\/strong><\/p>\n

We use Pipeline to chain multiple Transformers and Estimators together to specify our machine learning workflow. A Pipeline\u2019s stages are specified as an ordered array.<\/p>\n

from pyspark.ml import Pipeline<\/span>
\npipeline = Pipeline(stages = stages)<\/span>
\npipelineModel = pipeline.fit(df)<\/span>
\ndf = pipelineModel.transform(df)<\/span>
\nselectedCols = [‘label’, ‘features’] + cols<\/span>
\ndf = df.select(selectedCols)<\/span>
\ndf.printSchema()<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 7<\/p>\n

pd.DataFrame(df.take(5), columns=df.columns).transpose()<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 8<\/p>\n

As you can see, we now have features column and label column.<\/p>\n

Randomly split data into train and test sets, and set seed for reproducibility.<\/p>\n

train, test = df.randomSplit([0.7, 0.3], seed = 2018)<\/span>
\nprint(“Training Dataset Count: ” + str(train.count()))<\/span>
\nprint(“Test Dataset Count: ” + str(test.count()))<\/span><\/span><\/p>\n

Training Dataset Count: 7764
\nTest Dataset Count: 3398<\/em><\/strong><\/p>\n

Logistic Regression Model<\/h3>\n

from pyspark.ml.classification import LogisticRegression<\/span><\/span><\/p>\n

lr = LogisticRegression(featuresCol = ‘features’, labelCol = ‘label’, maxIter=10)<\/span>
\nlrModel = lr.fit(train)<\/span><\/span><\/p>\n

We can obtain the\u00a0coefficients<\/strong>\u00a0by using LogisticRegressionModel\u2019s attributes.<\/p>\n

import matplotlib.pyplot as plt<\/span>
\nimport numpy as np<\/span><\/span><\/p>\n

beta = np.sort(lrModel.coefficients)<\/span><\/span><\/p>\n

plt.plot(beta)<\/span>
\nplt.ylabel(‘Beta Coefficients’)<\/span>
\nplt.show()<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 9<\/p>\n

Summarize the model over the training set, we can also obtain the\u00a0receiver-operating characteristic and areaUnderROC<\/strong>.<\/p>\n

trainingSummary = lrModel.summary<\/span><\/span><\/p>\n

roc = trainingSummary.roc.toPandas()<\/span>
\nplt.plot(roc[‘FPR’],roc[‘TPR’])<\/span>
\nplt.ylabel(‘False Positive Rate’)<\/span>
\nplt.xlabel(‘True Positive Rate’)<\/span>
\nplt.title(‘ROC Curve’)<\/span>
\nplt.show()<\/span><\/span><\/p>\n

print(‘Training set areaUnderROC: ‘ + str(trainingSummary.areaUnderROC))<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 10<\/p>\n

Precision and recall<\/strong>.<\/p>\n

pr = trainingSummary.pr.toPandas()<\/span>
\nplt.plot(pr[‘recall’],pr[‘precision’])<\/span>
\nplt.ylabel(‘Precision’)<\/span>
\nplt.xlabel(‘Recall’)<\/span>
\nplt.show()<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 11<\/p>\n

Make predictions on the test set<\/strong>.<\/p>\n

predictions = lrModel.transform(test)<\/span>
\npredictions.select(‘age’, ‘job’, ‘label’, ‘rawPrediction’, ‘prediction’, ‘probability’).show(10)<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 12<\/p>\n

Evaluate our Logistic Regression model<\/strong>.<\/p>\n

from <\/span>pyspark<\/span>.ml.evaluation import BinaryClassificationEvaluator<\/span><\/span><\/p>\n

evaluator = BinaryClassificationEvaluator()<\/span>
\nprint(‘Test Area Under ROC’, evaluator.evaluate(predictions))<\/span><\/span><\/p>\n

Test Area Under ROC 0.8858324614449619<\/em><\/strong><\/p>\n

Decision Tree Classifier<\/strong><\/h3>\n

Decision trees are widely used since they are easy to interpret, handle categorical features, extend to the multi-class classification, do not require feature scaling, and are able to capture non-linearities and feature interactions.<\/p>\n

from pyspark.ml.classification import DecisionTreeClassifier<\/span><\/span><\/p>\n

dt = DecisionTreeClassifier(featuresCol = ‘features’, labelCol = ‘label’, maxDepth = 3)<\/span>
\ndtModel = dt.fit(train)<\/span>
\npredictions = dtModel.transform(test)<\/span>
\npredictions.select(‘age’, ‘job’, ‘label’, ‘rawPrediction’, ‘prediction’, ‘probability’).show(10)<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 13<\/p>\n

Evaluate our Decision Tree model<\/strong>.<\/p>\n

evaluator = BinaryClassificationEvaluator()<\/span>
\nprint(“Test Area Under ROC: ” + str(evaluator.evaluate(predictions, {evaluator.metricName: “areaUnderROC”})))<\/span><\/span><\/p>\n

Test Area Under ROC: 0.7807240050065357<\/em><\/strong><\/p>\n

One simple decision tree performed poorly because it is too weak given the range of different features. The prediction accuracy of decision trees can be improved by Ensemble methods, such as Random Forest and Gradient-Boosted Tree.<\/p>\n

Random Forest Classifier<\/strong><\/h3>\n

from pyspark.ml.classification import RandomForestClassifier<\/span><\/span><\/p>\n

rf = RandomForestClassifier(featuresCol = ‘features’, labelCol = ‘label’)<\/span>
\nrfModel = rf.fit(train)<\/span>
\npredictions = rfModel.transform(test)<\/span>
\npredictions.select(‘age’, ‘job’, ‘label’, ‘rawPrediction’, ‘prediction’, ‘probability’).show(10)<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 14<\/p>\n

Evaluate our Random Forest Classifier<\/strong>.<\/p>\n

evaluator = BinaryClassificationEvaluator()<\/span>
\nprint(“Test Area Under ROC: ” + str(evaluator.evaluate(predictions, {evaluator.metricName: “areaUnderROC”})))<\/span><\/span><\/p>\n

Test Area Under ROC: 0.8846453518867426<\/em><\/strong><\/p>\n

Gradient-Boosted Tree Classifier<\/strong><\/h3>\n

from pyspark.ml.classification import GBTClassifier<\/span><\/span><\/p>\n

gbt = GBTClassifier(maxIter=10)<\/span>
\ngbtModel = gbt.fit(train)<\/span>
\npredictions = gbtModel.transform(test)<\/span>
\npredictions.select(‘age’, ‘job’, ‘label’, ‘rawPrediction’, ‘prediction’, ‘probability’).show(10)<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 15<\/p>\n

Evaluate our Gradient-Boosted Tree Classifier.<\/strong><\/p>\n

evaluator = BinaryClassificationEvaluator()<\/span>
\nprint(“Test Area Under ROC: ” + str(evaluator.evaluate(predictions, {evaluator.metricName: “areaUnderROC”})))<\/span><\/span><\/p>\n

Test Area Under ROC: 0.8940728473145346<\/em><\/strong><\/p>\n

Gradient-Boosted Tree achieved the best results, we will try tuning this model with the ParamGridBuilder and the CrossValidator. Before that we can use explainParams() to print a list of all params and their definitions to understand what params available for tuning.<\/p>\n

print(gbt.explainParams())<\/span><\/span><\/p>\n

<\/canvas><\/figure>\n

Figure 16<\/p>\n

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator<\/span><\/span><\/p>\n

paramGrid = (ParamGridBuilder()<\/span>
\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .addGrid(gbt.maxDepth, [2, 4, 6])<\/span>
\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .addGrid(gbt.maxBins, [20, 60])<\/span>
\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .addGrid(gbt.maxIter, [10, 20])<\/span>
\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0 .build())<\/span><\/span><\/p>\n

cv = CrossValidator(estimator=gbt, estimatorParamMaps=paramGrid, evaluator=evaluator, numFolds=5)<\/span><\/span><\/p>\n

# Run cross validations.\u00a0 This can take about 6 minutes since it is training over 20 trees!<\/span>
\ncvModel = cv.fit(train)<\/span>
\npredictions = cvModel.transform(test)<\/span>
\nevaluator.evaluate(predictions)<\/span><\/span><\/p>\n

0.8981050997838095<\/em><\/strong><\/p>\n

To sum it up, we have learned how to build a binary classification application using PySpark and MLlib Pipelines API. We tried four algorithms and gradient boosting performed best on our data set.<\/p>\n

Source code can be found on\u00a0Github<\/a>.<\/p>\n

Reference: Apache Spark 2.1.0<\/a><\/p>\n","protected":false},"excerpt":{"rendered":"

Apache Spark is now becoming the big-data platform of choice for enterprises. It is a powerful open source engine that provides real-time stream processing, interactive processing, graph processing, in-memory processing as well as batch processing with very fast speed, ease of use and standard interface. Sooner or later, your company or your clients will be using Spark to develop sophisticated models that would enable you to discover new opportunities or avoid risk. Spark is not hard to learn, if you already know Python and SQL. learn how to build a binary classification application using PySpark and MLlib Pipelines API. <\/p>\n","protected":false},"author":255,"featured_media":14359,"comment_status":"open","ping_status":"open","sticky":false,"template":"single-post-2.php","format":"standard","meta":{"content-type":"","footnotes":""},"categories":[183],"tags":[97],"ppma_author":[2872],"class_list":["post-1519","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-ai-ml","tag-artificial-intelligence"],"authors":[{"term_id":2872,"user_id":255,"is_guest":0,"slug":"susan-li","display_name":"Susan Li","avatar_url":"https:\/\/secure.gravatar.com\/avatar\/?s=96&d=mm&r=g","user_url":"","last_name":"Li","first_name":"Susan","job_title":"","description":"Susan Li, Data Scientist at Wave Financial<\/a>, is helping organizations realize the potential of big data and advanced analytics.  Her specialities include Machine learning, data mining, and predictive modeling,  R, Python, SQL, and data visualization tools."}],"_links":{"self":[{"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/posts\/1519","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/users\/255"}],"replies":[{"embeddable":true,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/comments?post=1519"}],"version-history":[{"count":3,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/posts\/1519\/revisions"}],"predecessor-version":[{"id":30153,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/posts\/1519\/revisions\/30153"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/media\/14359"}],"wp:attachment":[{"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/media?parent=1519"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/categories?post=1519"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/tags?post=1519"},{"taxonomy":"author","embeddable":true,"href":"https:\/\/www.experfy.com\/blog\/wp-json\/wp\/v2\/ppma_author?post=1519"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}