[PySpark] Spark로 Encoding 과 Scaling 을 수행해보자!

🔊 해당 포스팅은 스파크 완벽 가이드 책과 인프런의 스파크 머신러닝 완벽 가이드 강의로 공부한 후 배운 내용을 저만의 방식으로 재구성한 것임을 알립니다. 특히, 참고한 인프런 강의의 강의 자료를 그대로 사용하지 않았음을 필히 알립니다!

Apache Spark를 Python으로 이용하는 PySpark에 대해 알아보자

 

---

이번 포스팅에서는 문자열 또는 Object 타입의 값을 숫자로 '인코딩' 해주는 기법들인 레이블 인코딩(Label Encoding)과 원-핫 인코딩(One-Hot Encoding)을 스파크로 수행하는 방법에 대해서 알아보도록 하자. 그리고 실수형 형태의 값들을 스케일링 해주는 방법에 대해서도 배워보자.

1. 문자열을 인덱스(숫자)로, StringToIndexer

사이킷런에 익숙한 분들은 레이블 인코딩을 수행한다고 하면 LabelEncoder 클래스나 OrdinalEncoder 클래스를 떠올리기 쉽다. 그런데 스파크에서는 해당 이름과 동일한 클래스는 존재하지 않고 StringToIndexer 라는 클래스가 존재한다. 이름 그대로 '문자열'인 String을 '숫자'인 Index로(to)를 의미한다. 사용법은 간단하다. 변환 대상인 문자열 타입의 칼럼 이름과 변환 후 만들어낼 칼럼 이름을 인자로 지정한 후, fit(), transform() 메소드를 사용해 아래처럼 변환할 수 있다.

 

from pyspark.ml.feature import StringIndexer

titanic_df = spark.read.csv('/FileStore/tables/train.csv', header=True, inferSchema=True)

# fit 하면 Model 객체가 반환됨. 그리고 transform을 수행해서 실질적인 변환 수행
indexer = StringIndexer(inputCol='Sex', outputCol='Sex_enc')
encoding_df = indexer.fit(titanic_df) \
                     .transform(titanic_df)
encoding_df.limit(5).show()

 

다음은 부가적으로 알아두면 좋을 클래스인데, 방금 레이블 인코딩 시킨 값을 다시 역으로 바꾸어줄 수 있는 클래스이다. 즉, '숫자' 에서 다시 '문자열'로 바꿀 수 있다. IndexToString 클래스를 사용하면 되는데, 신기한 점은 IndexToString 이라는 클래스를 별도로 임포트해서 사용하는 것인데, 이전에 StringToIndexer 할 때 수행했던 '문자열' - '숫자' 값을 알아서 찾아서 값을 복구 시켜준다.

 

from pyspark.ml.feature import IndexToString

stringer = IndexToString(inputCol='Sex_enc', outputCol='Sex_dec')
decoding_df = stringer.transform(encoding_df)
decoding_df.limit(3).show()

 

다음은 여러개의 칼럼들을 한 번에 레이블 인코딩시킬 수도 있다. 차이점이라고 한다면 여러개의 칼럼을 넣어줄 때는 리스트([]) 형태로 여러개의 칼럼이름을 아래처럼 넣어주면 된다. 참고로 인자 이름(inputCols, outputCols)을 보면 단일 칼럼일 때는 끝에 '-s'가 붙지 않지만 복수 칼럼일 때는 끝에 '-s'가 붙은 상태로 지정해주어야 한다.

 

from pyspark.ml.feature import StringIndexer

indexer = StringIndexer(inputCols=['Sex', 'Embarked'], outputCols=['Sex_enc', 'Embarked_enc'])
encoding_df2 = indexer.fit(titanic_df) \
                      .transform(titanic_df)
encoding_df2.limit(3).show()

 

하지만 다시 값을 복구시키는 IndexToString는 다르게 한번에 여러개의 칼럼을 복구시킬 수는 없다. 이 점이 약간 아쉬운데 여러개의 칼럼을 복구시키려면 일일이 하나씩 아래처럼 해주어야 한다.

 

from pyspark.ml.feature import IndexToString

sex_stringer = IndexToString(inputCol='Sex_enc', outputCol='Sex_dec')
embarked_stringer = IndexToString(inputCol='Embarked_enc', outputCol='Embarked_dec')

decoding_df2 = sex_stringer.transform(encoding_df2)
decoding_df2 = embarked_stringer.transform(decoding_df2)
decoding_df2.limit(5).show()

2. 문자열을 One-Hot 벡터로 변환시키자, OneHotEncoder

문자열의 값을 원-핫 벡터로 변환시킬 때 사용하는 클래스는 스파크에서도 사이킷런과 똑같이 OneHotEncoder 클래스를 사용한다. 하지만 사이킷런과 다르게 차이점이 존재한다. 사이킷런은 OneHotEncoder를 사용할 때 문자열 형태의 칼럼을 그대로 넣어주면 알아서 내부적으로 레이블 인코딩을 호출해 수행한 후, 원-핫 벡터로 변환시켜준다.

 

하지만 스파크의 OneHotEncoder는 그렇게 동작하지 않는다. 그래서 반드시 직접 문자열에서 숫자로 변환해주는 레이블 인코딩 작업을 사전에 수행해주어야 한다. 아래 코드를 보면서 이해해보도록 하자.

 

# 원-핫 인코딩 수행하려먼 먼저 레이블인코딩을 수행해야 함!
from pyspark.ml.feature import StringIndexer
from pyspark.ml.feature import OneHotEncoder

indexer = StringIndexer(inputCols=['Sex', 'Embarked'], outputCols=['Sex_label', 'Embarked_label'])
label_df = indexer.fit(titanic_df).transform(titanic_df)

ohe_encoder = OneHotEncoder(inputCols=['Sex_label', 'Embarked_label'], outputCols=['Sex_ohe', 'Embarked_ohe'], dropLast=True)
encoding_df = ohe_encoder.fit(label_df).transform(label_df)

encoding_df.limit(5).show()

 

위에서 헷갈리지 말아야 할 점은 OneHotEncoder 클래스를 사용할 때 dropLast 라는 인자이다. dropLast 를 True로 설정하면, 원-핫 벡터로 변환 시 마지막 벡터를 드랍을 하게 된다. 이게 무슨 말일까? 우선 아래와 같은 범주형 변수가 있다고 가정해보자.

 

3종류의 범주형 값이 있다

위 그림처럼 $x$ 변수는 A, B, C 총 3종류의 값을 갖는다. 원래대로라면 길이가 3인 원-핫 벡터로 나타낼 수 있다(drop_last=False인 경우의 그림) 하지만 길이가 2인 원-핫 벡터로 3종류의 값을 나타낼 수도 있다. 잘 생각해보면 모든 원소가 0일 때도 하나의 종류값을 나타낼 수 있기 때문이다. 그래서 drop_last=True인 경우의 그림을 보면 길이가 2인 벡터로 A, B, C 총 3개의 값을 나타낼 수 있는 것이다.

3. 수치형 변수를 스케일링하기!

이번엔 Standard Scaling, Min-Max Scaling 등 다양한 스케일링 기법을 활용해서 스파크에서 수치형 값들을 스케일링 하는 방법에 대해서 알아보도록 하자. 스파크에서는 특이하게 스케일링을 적용하는 칼럼은 반드시 Feature Vectorization된 칼럼이어야 한다. Feature Vectorization이 무엇인지에 대해서는 예전 스파크 포스팅에서 배운 적이 있다.

 

따라서 스케일링을 적용하기 전에 반드시 VectorAssembler 를 활용해 Feature Vectorization 시킨 후에 적용해야 한다. 붓꽃 데이터를 활용해서 알아보자.

 

from sklearn.datasets import load_iris
import pandas as pd
import numpy as np
from pyspark.ml.feature import StandardScaler
from pyspark.ml.feature import VectorAssembler

iris = load_iris()
iris_columns = ['sepal_length', 'sepal_width', 'petal_length', 'petal_width']
iris_df = pd.DataFrame(iris.data, columns=iris_columns)
iris_df['label'] = iris.target

iris_sdf = spark.createDataFrame(iris_df)

vec_assembler = VectorAssembler(inputCols=iris_sdf.columns[:-1], outputCol='feature')
iris_vec = vec_assembler.transform(iris_sdf)

scaler = StandardScaler(inputCol='feature', outputCol='feature_scaled', withMean=True, withStd=True)  # True로 하면 평균은 0, std=1로 '표준정규분포'
iris_scaled = scaler.fit(iris_vec).transform(iris_vec)
iris_scaled.limit(5).show()

 

참고로 StandardScaler를 사용할 때, withMean 이라는 인자가 기본적으로 False로 설정된다. withMean, withStd 인자를 True로 설정한다면, 평균이 0, 표준편차가 1인 즉, 표준정규분포 형태로 값을 스케일링해줌을 의미한다. 만약 False로 설정한다면 스케일링을 적용할 수치형 값들의 실제 평균값으로 설정되어 정규화가 수행된다는 점도 알아두자.