[PySpark] Spark의 Pipeline으로 분류 모델을 만들어보자!

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

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

---

이번 포스팅에서는 PySpark를 활용해서 간단한 머신러닝 분류 모델을 만들어보도록 하자. 그리고 분류 모델을 만드는 데 거치는 여러가지 과정을 하나의 파이프라인으로 만들 수 있는 PySpark의 Pipeline을 활용해보는 방법도 알아보도록 하자.

1. PySpark로 간단한 Multi-Class 분류 모델 만들기

해당 포스팅에서는 대표적인 벤치마크 데이터인 IRIS(붓꽃) 데이터를 활용한 멀티-클래스 분류 모델을 만들어보자. 우선 아래의 Scikit-learn 모듈을 활용해서 붓꽃 데이터를 로드하고 스파크 데이터프레임으로 변환시키자.

 

from sklearn.datasets import load_iris
import pandas as pd
import numpy as np

iris = load_iris()
data = iris.data
label = iris.target

iris_df = pd.DataFrame(data, columns=iris.feature_names)
iris_df['target'] = label

spark_df = spark.createDataFrame(iris_df)

1-1. 독립변수들을 하나의 벡터로, Feature Vectorization!

Feature Vectorization은 스파크 데이터프레임에 있는 하나 또는 둘 이상의 독립변수(Feature)들을 하나의 벡터로 묶는 것을 의미한다. 그림으로 표시하면 아래와 같다.

 

Feature Vector를 만드는 것이 Feature Vectorization 이다!

 

위 그림을 보면 Feature Vectorization 과정이 어떤 것을 의미하는지 이해할 수 있을 것이다. 위 과정을 스파크로 구현하려면 스파크의 ML 모듈에서 제공하는 VectorAssembler 클래스를 사용하면 된다.

 

from pyspark.ml.feature import VectorAssembler

# 1. Train/Test 데이터로 분할
train_df, test_df = spark_df.randomSplit(weights=[0.8, 0.2], seed=42)

# 2. Feature Vector로 만들기
ftr_columns = train_df.columns[:-1]
vec_assembler = VectorAssembler(inputCols=ftr_columns, outputCol='features')

train_ftr_vec = vec_assembler.transform(train_df)
test_ftr_vec = vec_assembler.transform(test_df)  # 테스트 데이터에 대해도 동일한 객체 사용

print('# 학습 원본 데이터:')
train_df.limit(5).show()

print('# Feature Vectorization 후 학습 데이터:')
train_ftr_vec.limit(5).show()

 

VectorAssembler 클래스를 사용할 때는 inputCols 인자에 독립변수들 칼럼 이름들을 리스트 형태로 넣어주면 된다. outputCol 에는 독립변수들을 하나의 벡터화 즉, Feature Vectorization 시켰을 때 만들어질 하나의 칼럼에 대해 명시해줄 칼럼 이름이다.

 

위 코드에서는 'features' 라는 문자열로 명시해주었는데, 아래 코드 결과를 보면 features 라는 새로운 칼럼이 생긴 것을 알 수 있다. 이렇게 하나의 VectorAssembler 객체를 생성해준 후, 학습 데이터에 대해 transform 메소드로 실질적인 Feature Vectorization을 수행해주면 된다.

 

단, 검증(또는 테스트) 데이터에 대해서도 Feature Vectorization을 적용해줄 때 학습 데이터에 적용할 때 만든 동일한 VectorAssembler 객체를 사용하는 것이 좋다.

 

위 코드 결과인 두 VectorAssembler 적용 전/후 데이터프레임의 결과 차이는 아래와 같다.

 

VectorAssembler 적용 전/후의 차이

1-2. 멀티-클래스 분류 모델로 학습, 예측하기

해당 예시에서는 사용할 분류 모델로서 의사결정 나무(Decision Tree)를 사용하기로 하자. 사이킷런 때와 비슷한 API 이지만 차이점이 어느정도 있다. 먼저 아래의 코드를 수행해서 모델을 정의하고 학습 데이터에 대해 모델을 학습시키고 예측하는 코드를 살펴보자.

 

from pyspark.ml.classification import DecisionTreeClassifier

# 모델 정의
dt_clf = DecisionTreeClassifier(featuresCol='features', labelCol='target', maxDepth=10)

# 학습
dt_model = dt_clf.fit(train_ftr_vec)

# 학습, 테스트 데이터에 대해 예측
train_pred = dt_model.transform(train_ftr_vec)
test_pred = dt_model.transform(test_ftr_vec)

 

먼저 모델을 정의할 때, 사용할 모델의 클래스를 사용해서 featuresCol 에는 [1-1. 목차]에서 만든 Feature Vector의 칼럼 이름을 명시해준다. 그리고 labelCols 인자에는 우리가 예측하려는 '정답' 즉, 종속변수 칼럼의 이름을 명시해준다. 그리고 추가적으로 사용하려는 머신러닝 모델의 하이퍼파라미터 인자를 추가해준다. 여기서는 maxDepth 라는 Decision Tree 모델의 하이퍼파라미터를 명시했다.

 

그리고 학습 시, Feature Vectorization 시켜준 데이터프레임을 fit() 메소드에 넣어준다. 이 때, fit() 메소드를 호출할 시, Model 클래스 객체가 반환된다. 이것이 아마 사이킷런과의 가장 API 적 차이점이라고 할 수 있겠다.

 

그리고 학습, 테스트 데이터에 대해 모델의 예측값을 내뱉기 위해서는 fit() 메소드로 반환된 Model 클래스 객체에서 transform() 이라는 메소드를 활용한다. transform() 메소드 인자에는 예측하려는 Feature Vectorization이 적용된 데이터프레임을 넣어주면 된다. 이렇게 예측함으로써 반환된 객체들(위 코드 상에서는 train_pred, test_pred 변수)은 스파크의 데이터프레임 객체로 이것들의 결과를 출력해보면 아래 사진처럼 여러가지 칼럼들이 있는 것을 알 수 있다.

 

학습 데이터에 대한 예측값 데이터프레임

 

위 사진을 보면 알다시피 원본 Feature, Feature Vectorization된 것, 모델이 예측한 Raw한 결과값, 그리고 Raw한 결과값에 활성함수를 적용해 반환된 확률 값, 최종 예측값까지 담겨있는 것을 볼 수 있다. 우리는 밑에서 모델 성능을 평가하기 위해서 prediction 이라는 칼럼에 있는 값들을 활용하면 된다.

1-3. 모델의 성능 평가하기

해당 예시에서 해결하려는 문제는 멀티-클래스 분류 문제이기 때문에 이에 맞는 Evaluator 클래스를 활용해야 한다.

 

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

evaluator = MulticlassClassificationEvaluator(predictionCol='prediction', labelCol='target', metricName='accuracy')

train_acc = evaluator.evaluate(train_pred)
test_acc = evaluator.evaluate(test_pred)

print('Train Accuracy:', train_acc)
print('Test Accuracy:', test_acc)

 

참고로 스파크에는 멀티-클래스가 아닌 멀티-레이블 문제에 대해 다루는 MultilabelClassificationEvaluator 도 존재한다. 멀티-클래스, 멀티-레이블의 차이점은 무엇인지 예전 포스팅에서도 언급했었다. 클래스가 여러개인 것 중 무조건 하나만을 가져야 한다면 멀티-클래스, 동시에 여러개의 클래스를 가져도 된다면 멀티-레이블이다.

 

2. Pipeline을 활용해 전처리, 모델링 과정을 간소화시키자!

다음은 스파크에서 제공하는 Pipeline 이라는 클래스를 활용해서 데이터 전처리, Feature Vectorization, 모델링 과정을 간소화시킬 수 있는 방법에 대해 배워보자. 사이킷런도 Pipeline이라는 클래스를 활용할 수 있는데, 예전에 사용하면서 편리했던 경험이 있다. 이러한 경험을 스파크에서도 할 수 있다!

 

먼저 스파크 데이터프레임을 Train, Test 용 데이터로 분할하고 Feature Vectorization을 수행하는 VectorAssembler 를 하나의 단계(stage)로 정의하고, 또 하나는 모델을 정의함으로써 또다른 단계(stage)를 정의해보자.

 

from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import DecisionTreeClassifier

# 1. 데이터 분할
train_df, test_df = spark_df.randomSplit(weights=[0.8, 0.2], seed=42)

# 2. Feature Vector 데이터프레임으로 변환하는 스테이지 생성
ftr_columns = train_df.columns[:-1]
stage_1 = VectorAssembler(inputCols=ftr_columns, outputCol='features')

# 3. 모델 정의를 하나의 스테이지로 생성
stage_2 = DecisionTreeClassifier(featuresCol='features', 
                                 labelCol='target')

 

이제 스파크에서 제공하는 Pipeline 클래스를 활용해서 위에서 만든 stage들을 하나의 파이프라인으로 정의해 한꺼번에 실행할 수 있도록 만들어보자.

 

from pyspark.ml import Pipeline

# 4. 두 스테이지를 하나의 파이프라인으로 만들기
pipeline = Pipeline(stages=[stage_1, stage_2])

# 5. 파이프라인을 fit 하면 각 스테이지에서 transform 또는 fit 메소드를 학습 데이터에 수행
pipeline_model = pipeline.fit(train_df)

# 6. 5번 단계로 만든 파이프라인 모델을 transform으로 학습 및 테스트 데이터에 수행
train_pred = pipeline_model.transform(train_df)
test_pred = pipeline_model.transform(test_df)

 

정의된 파이프라인 객체에서 fit() 메소드를 수행하게 되면 정의한 각 stage들, 여기에서는 VectorAssembler 객체의 transform() 메소드와 모델 객체에서의 fit() 메소드를 순차적으로 자동으로 실행시켜준다. 즉, 학습 데이터에 대해 데이터 전처리, 모델 학습이 수행되는 것이다.

 

헷갈리지 말아야 할 점은 파이프라인 객체에서는 fit() 메소드를 수행하니까 'fit' 이라는 메소드 이름 자체만 보고 "왜 VectorAssembler 객체에서는 transform() 을 수행하지?" 라고 생각할 수 있다. 하지만 스파크를 제작한 개발자 분들이 파이프라인 객체서 fit() 메소드를 수행하면 VectorAssembler 객체에서는 fit() 이 아닌 transform() 메소드를 수행할 수 있도록 로직을 작성해 두었다.(참고로 VectorAssembler 에는 fit() 메소드 자체가 존재하지 않다는 점도 있다)

 

이렇게 파이프라인 객체에서 fit() 메소드로 파이프라인 모델 객체를 반환받는다. 그리고 난 후, 파이프라인 모델 객체에서 transform() 메소드를 활용해 테스트 데이터에 대해 예측을 수행할 수 있다. 이렇게 반한된 학습, 테스트 데이터에 대한 예측값 결과 화면은 아래와 같다.

 

파이프라인 모델의 transform() 결과 화면

 

이렇게 파이프라인을 활용해서 쉽게 데이터 전처리 및 모델 학습, 예측하는 과정을 하나의 과정으로 간결하게 처리할 수 있다. 물론 해당 예시에서는 스케일링이나 범주형 변수에 대한 레이블 또는 원-핫 인코딩 과정이 존재하지는 않지만, 만약 이러한 전처리 과정이 추가로 필요하다면 위에서 했던 것처럼 전처리 순서대로 stage들을 정의해주고 순서에 맞게 리스트 형태로 정의한 후 파이프라인 객체로 만들어주면 된다.

 

참고로 파이프라인 모델 객체를 dir() 매직 메소드로 어떤 속성값들이 있는지 살펴보면 여러가지 속성값들이 있는 것을 알 수 있다. 그 중에 stages라는 속성 값이 아래처럼 있는데, 이 속성 값을 통해서 원하는 특정 단계를 따로 추출해서 활용할 수도 있다.

 

파이프라인 모델 객체의 여러가지 속성값