Feature Transformer VectorAssembler in PySpark ML Feature — Part 3

What is VectorAssembler?

class pyspark.ml.feature.VectorAssembler(inputCols=None, outputCol=None, handleInvalid=’error’):

VectorAssembler is a transformer that combines a given list of columns into a single vector column.

It is useful for combining raw features and features generated by different feature transformers into a single feature vector, in order to train ML models like logistic regression and decision trees.

VectorAssembler accepts the following input column types: all numeric types, boolean type, and vector type. In each row, the values of the input columns will be concatenated into a vector in the specified order.

Note: For VectorAssembler, we do not need StringIndexer and OneHotEncoder, if your data have all numeric values. In this example we have string columns, so we are using StringIndexer and OneHotEncoder.

Part 1 — What is StringIndexer?

We have already discussed regarding StringIndexer (link)

Part 2 — What is OneHotEncoder?

We have already discussed regarding OneHotEncoder (link)

Let us see an example

Create SparkSession

#import SparkSession
from pyspark.sql import SparkSession

SparkSession is an entry point to Spark to work with RDD, DataFrame, and Dataset. To create SparkSession in Python, we need to use the builder() method and calling getOrCreate() method.

If SparkSession already exists it returns otherwise create a new SparkSession.

spark = SparkSession.builder.appName('xvspark').getOrCreate()

Create dataframe by declaring the schema

from pyspark.sql.types import *

StructType class to define the structure of the DataFrame.

#create the structure of schema
schema = StructType().add("id","integer").add("name","string").add("qualification","string").add("age", "integer").add("gender", "string").add("passed", "integer")
#create data
data = [
    (1,'John',"B.A.", 20, "Male", 1),
    (2,'Martha',"B.Com.", 20, "Female", 1),
    (3,'Mona',"B.Com.", 21, "Female", 1),
    (4,'Harish',"B.Sc.", 22, "Male", 1),
    (5,'Jonny',"B.A.", 22, "Male", 0),
    (6,'Maria',"B.A.", 23, "Female", 1),
    (7,'Monalisa',"B.A.", 21, "Female", 0)
]

#create dataframe
df = spark.createDataFrame(data, schema=schema)
#columns of dataframe
df.columns

Output:

[‘id’, ‘name’, ‘qualification’, ‘age’, ‘gender’, ‘passed’]

df.show()

Output:

Apply StringIndexer & OneHotEncoder to qualification and gender columns

#import required libraries
from pyspark.ml.feature import StringIndexer

Apply StringIndexer to qualification column

qualification_indexer = StringIndexer(inputCol="qualification", outputCol="qualificationIndex")
#Fits a model to the input dataset with optional parameters.
df = qualification_indexer.fit(df).transform(df)
df.show()

Output:

“B.A.” gets index 0 because it is the most frequent, then “B.Com” gets index 1 and “B.Sc.” gets index 2.

Apply StringIndexer to gender column

gender_indexer = StringIndexer(inputCol="gender", outputCol="genderIndex")
#Fits a model to the input dataset with optional parameters.
df = gender_indexer.fit(df).transform(df)
df.show()

Output:

Apply OneHotEncoder to qualificationIndex column

from pyspark.ml.feature import OneHotEncoder

#onehotencoder to qualificationIndex
onehotencoder_qualification_vector = OneHotEncoder(inputCol="qualificationIndex", outputCol="qualification_vec")
df = onehotencoder_qualification_vector.fit(df).transform(df)
df.show()

Output:

Apply OneHotEncoder to genderIndex column

#onehotencoder to genderIndex
onehotencoder_gender_vector = OneHotEncoder(inputCol="genderIndex", outputCol="gender_vec")
df = onehotencoder_gender_vector.fit(df).transform(df)
df.show()

Output:

Feature transformer — VectorAssembler

We want to combine age, qualification_vec, and gender_vec into a single feature vector called features and use it to predict passed or not.

If we set VectorAssembler’s input columns to age, qualification_vec, and gender_vec and output column to features.

from pyspark.ml.feature import VectorAssembler

#dataframe columns 
df.columns

Output:

[‘id’,
‘name’,
‘qualification’,
‘age’,
‘gender’,
‘passed’,
‘qualificationIndex’,
‘genderIndex’,
‘qualification_vec’,
‘gender_vec’]

inputCols = [
 'age',
 'qualification_vec',
 'gender_vec'
]
outputCol = "features"
df_va = VectorAssembler(inputCols = inputCols, outputCol = outputCol)
df = df_va.transform(df)
df.select(['features']).toPandas().head(5)

Output:

new_df = df.select(['features','passed'])
new_df.show()

Output:

Using Pipeline

#import module
from pyspark.ml import Pipeline

Reload Data

#create the structure of schema
schema = StructType().add("id","integer").add("name","string").add("qualification","string").add("age", "integer").add("gender", "string").add("passed", "integer")

#create data
data = [
    (1,'John',"B.A.", 20, "Male", 1),
    (2,'Martha',"B.Com.", 20, "Female", 1),
    (3,'Mona',"B.Com.", 21, "Female", 1),
    (4,'Harish',"B.Sc.", 22, "Male", 1),
    (5,'Jonny',"B.A.", 22, "Male", 0),
    (6,'Maria',"B.A.", 23, "Female", 1),
    (7,'Monalisa',"B.A.", 21, "Female", 0)
]
df = spark.createDataFrame(data, schema=schema)
df.show()

Output:

Create Pipeline and pass all stages

#Convert qualification and gender columns to numeric
qualification_indexer = StringIndexer(inputCol="qualification", outputCol="qualificationIndex")
gender_indexer = StringIndexer(inputCol="gender", outputCol="genderIndex")

#Convert qualificationIndex and genderIndex
onehot_encoder = OneHotEncoder(inputCols=["qualificationIndex", "genderIndex"], outputCols=["qualification_vec", "gender_vec"])

#Merge multiple columns into a vector column
vector_assembler = VectorAssembler(inputCols=['age', 'qualification_vec', 'gender_vec'], outputCol='features')
#Create pipeline and pass it to stages
pipeline = Pipeline(stages=[
           qualification_indexer, 
           gender_indexer,
           onehot_encoder,
           vector_assembler
])
#fit and transform
df_transformed = pipeline.fit(df).transform(df)
df_transformed.show()

Output:

df_transformed = df_transformed.select(['features','passed'])
df_transformed.show()

Output:

You can convert it to Pandas DataFrame

df_transformed.toPandas()

Output: