Implementing a Machine Learning Pipeline Using PySpark Library

Dealing with Categorical Columns

As we discussed dealing with the categorical columns which are now in the String state, but as we know, String type is not accepted by any ML algorithm, so we need to deal with it, and for that, we have to go through a set of operations/steps in PySpark.

So, let’s break down each step and convert the necessary features columns.

Vector Assembler and OneHotEncoder

Vector Assembler: From the name itself it is indicating that it kind of puts together columns in a collective vectorized format i.e., all the features get stacked up as a single unit in the form of a vector, and this is one of the rules as well that MLIB library takes all the features as a single unit only.

One Hot Encoder: There are multiple ways of dealing with categorical variables this time going with One hot encoder where each categorical value is separated into an independent column and gets the binary value i.e. either 0 or 1.

from pyspark.ml.feature import (VectorAssembler,VectorIndexer,
                                OneHotEncoder,StringIndexer)

gender_indexer = StringIndexer(inputCol='Sex',outputCol='SexIndex')
gender_encoder = OneHotEncoder(inputCol='SexIndex',outputCol='SexVec')

embark_indexer = StringIndexer(inputCol='Embarked',outputCol='EmbarkIndex')
embark_encoder = OneHotEncoder(inputCol='EmbarkIndex',outputCol='EmbarkVec')

assembler = VectorAssembler(inputCols=['Pclass',
 'SexVec',
 'Age',
 'SibSp',
 'Parch',
 'Fare',
 'EmbarkVec'],outputCol='features')

Code breakdown:

1. As discussed, the Vector assembler and One Hot encoding technique are required for conversion; hence we imported both of them from the ml. feature library of PySpark.
2. While importing other important methods, note that StringIndexer was also there, which would be responsible for converting the String type to a categorical type.
3. Then One Hot Encoder will convert each categorical value to its binary value, i.e., 0 or 1, by its predefined object. Repeating the same process for the “Embarked” column as we did for the “Gender” column.
4. At last, Vector Assembler will put together all the preprocessed feature columns together and remove the unwanted ones.

from pyspark.ml.classification import LogisticRegression

So we are good to go with the model development phase, and the first thing we need to import is the ML algorithm; for this particular problem statement, we have to predict the categorical data; hence classification machine learning algorithm should be accessed, i.e.,, Logistic Regression.

Pipelines from PySpark

Sometimes coping with the whole process of model development is complex. We get stuck to choosing the right flow if the execution in this type of problem Pipelines from PySpark comes in to rescue us as it helps maintain the execution cycle flow so that each step should be performed at its best given stage neither before nor soon.

from pyspark.ml import Pipeline

log_reg_titanic = LogisticRegression(featuresCol='features',labelCol='Survived')

pipeline = Pipeline(stages=[gender_indexer,embark_indexer,
                           gender_encoder,embark_encoder,
                           assembler,log_reg_titanic])

train_titanic_data, test_titanic_data = my_final_data.randomSplit([0.7,.3])

fit_model = pipeline.fit(train_titanic_data)

results = fit_model.transform(test_titanic_data)

Code breakdown:

1. First and foremost Pipeline module is being accessed and imported by the pyspark. ml library.
2. Then for developing the model, the Logistic Regression method is used in the parameters passing in the features columns and label (independent) column.
3. Now comes the Pipeline method, where one can look in the stages section that all the preprocessed steps are lined up one after the other.
4. Then using the random split() method, the final dataset is broken down into a training set of 70% and a testing set of 30%.
5. At last, it’s important to have all the changes committed for that we are first fitting the pipeline with training data and transforming the testing data with the pipeline model.

Model Evaluation

Okay! So we are now in the Model Evaluation phase which means development is already done and now we should evaluate it from evaluating, we mean that it should be working as per our requirement with good results i.e., good accuracy over testing data.

from pyspark.ml.evaluation import BinaryClassificationEvaluator

my_eval = BinaryClassificationEvaluator(rawPredictionCol='prediction',
                                       labelCol='Survived')

results.select('Survived','prediction').show()

Output:

Code breakdown:

1. Imported the Binary Classifier as this problem statement has the binary type of target column.
2. Then by applying the Binary Classification evaluator object, we are passing in the values to the raw prediction column and the label column
3. At the end, when the DataFrame has both Survived and Prediction columns after the evaluation, then it is shown using a select statement.

Conclusion

Here we come in the final section of this article, where we will allow ourselves to go along with whatever we did so far in this article, i.e. from starting the spark session to building and evaluating the model, we will discuss each step in brief.

1. Firstly we started the spark session and read the famous titanic survivals dataset using PySpark data preprocessing techniques.
2. Then, we dealt with NULL values by dropping all of them. Along with that, we also handled the categorical features and converted them to relevant types using Vector Assembler and One Hot Encoder.
3. During the next phase, we came across the concept of Pipelines which helped us to build an end-to-end pipeline of all the stages.
4. At last, we build the Logistic regression model using PySpark MLIB and later evaluate it too so that we should see how well our model performed based on the testing data.

Here’s the repo link to this article. I hope you liked my article on A guide to implementing a machine learning pipeline using PySpark Library. If you have any opinions or questions, then comment below.

Connect with me on LinkedIn for further discussion.