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.
We have already discussed regarding StringIndexer (link)
We have already discussed regarding OneHotEncoder (link)
#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()
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
df.show()
#import required libraries
from pyspark.ml.feature import StringIndexer
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()
"B.A." gets index 0 because it is the most frequent, then "B.Com" gets index 1 and "B.Sc." gets index 2.
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()
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()
#onehotencoder to genderIndex
onehotencoder_gender_vector = OneHotEncoder(inputCol="genderIndex", outputCol="gender_vec")
df = onehotencoder_gender_vector.fit(df).transform(df)
df.show()
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
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)
new_df = df.select(['features','passed'])
new_df.show()
#import module
from pyspark.ml import Pipeline
#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()
#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()
df_transformed = df_transformed.select(['features','passed'])
df_transformed.show()
df_transformed.toPandas()