[PySpark] Spark로 교차검증, 파라미터 튜닝을 해보자!

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

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

---

이번 포스팅에서는 스파크를 활용해서 모델의 교차검증을 수행하고 모델의 하이퍼파라미터를 튜닝을 함께 수행하는 방법에 대해 알아보자. 그리고 더 나아가 교차검증과 파라미터 튜닝 과정을 저번 포스팅에서 배운 Pipeline을 활용해 수행하는 방법도 알아보도록 하자.

1. 교차검증과 파라미터 튜닝을 동시에, CrossValidator

스파크의 CrossValidator 클래스는 교차검증과 파라미터 튜닝을 동시에 수행할 수 있도록 한다. 여기서 '교차검증'이란 K-fold 교차검증을 의미한다. K-fold 교차검증에 대해 모른다면 예전 포스팅을 참고하도록 하자.

 

그리고 파라미터 튜닝은 파라미터 조합의 '모든 경우의 수'를 탐색하는 그리드 서치(Grid Search) 방법을 활용한다. 참고로 교차검증을 수행할 때, (분류 문제의 경우)클래스의 불균형을 고려한 Stratified K-fold 교차검증을 수행하지는 않는다. 즉, 스파크의 CrossValidator 는 클래스 개수 분포를 고려하지 않고 그냥 무작위로 학습/검증 데이터를 분할한다는 한계점은 존재한다.

 

CrossValidator 클래스는 인자로 크게 사용할 모델 객체(Estimator), 튜닝할 파라미터 범위(이를 Parameter Grid 라고 보통 일컫는다), 교차검증을 수행하면서 모델의 성능을 평가할 Evaluator, 교차검증할 때 사용할 Fold 개수 총 4개를 넣어준다. 그래서 CrossValidator 를 실질적으로 사용하기 위해서는 방금 언급한 4개의 인자에 넣어줄 객체 또는 값을 선언해주어야 한다. 이제 코드를 통해 차례차례 살펴보도록 하자.

 

사용하는 데이터는 붓꽃(iris) 데이터이다. 먼저 Scikit-learn을 활용해 pandas.dataframe 형태로 데이터를 로드하고 이를 spark.dataframe 객체로 변환시키자.

 

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

iris = load_iris()
iris_data = iris.data
iris_label = iris.target

iris_columns = ['sepal_length', 'sepal_width', 'petal_length', 'petal_width']
iris_pdf = pd.DataFrame(iris_data, columns=iris_columns)
iris_pdf['label'] = iris_label

iris_sdf = spark.createDataFrame(iris_pdf)

 

다음은 로드한 붓꽃 데이터를 Train, Test 용으로 나누고 변수들은 Feature Vectorization을 시키자. 그리고 사용할 모델인 Decision Tree를 정의해보자.

 

# Feature Vectorization 변환과 모델 estimator 객체 생성
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.classification import DecisionTreeClassifier
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

feature_cols = iris_sdf.columns[:-1]

# 데이터 분할
train, test = iris_sdf.randomSplit(weights=[0.7, 0.3], seed=42)
# Feature Vectorization
vec_assembler = VectorAssembler(inputCols=feature_cols, 
                                outputCol='feature')
train_vec = vec_assembler.transform(train)
test_vec = vec_assembler.transform(test)

# 모델 정의
dt_clf = DecisionTreeClassifier(featuresCol='feature', labelCol='label')

 

이제 다음으로는 이번 포스팅에서 핵심인 튜닝할 파라미터의 범위를 정의해주도록 하자. 이 때, 파라미터 범위를 지정할 때는 방금 위에서 정의한 Decision Tree 모델 객체의 속성값을 이어받도록 한다.

 

from pyspark.ml.tuning import ParamGridBuilder

# 파라미터 튜닝할 범위를 지정하는 param grid
param_grid = ParamGridBuilder().addGrid(dt_clf.maxDepth, [5, 20]) \
                               .addGrid(dt_clf.minInstancesPerNode, [2, 50])\
                               .build()

 

위 코드를 보면 약간 문법 형태가 이상하다고 느낄 수도 있다. Functional API 형태로 작성되었는데, Tensorflow 2.x 버전 문법에 익숙하신 분들은 아마 손쉽게 이해할 수 있을 것이다. 위와 같은 방식을 활용해서 사용할 모델의 파라미터 범위를 지정해주자.

 

다음으로는 교차검증을 수행할 때 모델의 성능을 평가할 Evaluator를 정의해주고 실질적인 '교차검증'을 수행할 클래스인 CrossValidator 클래스를 정의하자.

 

from pyspark.ml.tuning import CrossValidator

# 파라미터 튜닝하면서 측정할 메트릭 evaluator
evaluator = MulticlassClassificationEvaluator(labelCol='label',
                                             predictionCol='prediction',
                                             metricName='accuracy')
# Cross Validator 객체 생성 -> [모델, Param grid, evaluator, 폴드 수] 인자로 넣기
cv = CrossValidator(estimator=dt_clf,
                    estimatorParamMaps=param_grid,
                   evaluator=evaluator,
                   numFolds=3)

​

위에서 CrossValidator 클래스를 활용해 정의한 객체를 사용해서 학습 데이터에 대해 학습, 파라미터 튜닝, 교차검증을 수행시켜보자. fit() 메소드를 활용하면 된다.

 

cv_model = cv.fit(train_vec)

​

참고로 위의 fit() 메소드를 활용하게 되면 파라미터 튜닝을 수행한 후, 가장 최고의 성능에 도달한 파라미터를 기준으로 마지막에 재학습(refit)한 후의 모델 객체를 반환하게 된다. 그래서 이 모델 객체(위 코드 상으로는 cv_model 이라는 변수)의 transform() 메소드를 활용해서 테스트 데이터에 대해 최종 예측을 수행할 수 있게 된다.

 

from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# 테스트 데이터에 대해서 cv_model 객체의 transform 메소드를 활용
prediction = cv_model.transform(test_vec)  # spark.dataframe

# 테스트용 evaluator
test_evaluator = MulticlassClassificationEvaluator(labelCol='label',
                                                  predictionCol='prediction',
                                                  metricName='accuracy')
test_acc = test_evaluator.evaluate(prediction)
print('Test_acc:', round(test_acc, 3))

2. 가장 마지막의 데이터만 검증 데이터로 사용하자, TrainValidationSplit

모델의 교차검증을 수행하는 또 다른 방법으로 TrainValidationSplit 클래스가 존재한다. TrainValidationSplit는 CrossValidator 와는 달리 전체 데이터 중 특정 비율의 가장 마지막 순서에 있는 데이터만 검증 데이터로 활용하는 것이다. 두 가지의 차이점을 알아보기 위한 도식화 그림은 아래와 같다.

 

CrossValidator 와 TrainValidationSplit 검증 수행 방법의 차이

 

위 그림을 보면 자연스럽게 검증을 수행하는 횟수에서도 차이가 나는 것을 알 수가 있다. 왼쪽은 5번 수행하는 반면, 오른쪽은 1번 밖에 수행하지 않는다. 그렇다면 굳이 오른쪽의 방법을 이용할 필요가 있을까? 오른쪽 방법을 이용하는 경우를 들자면, 튜닝할 파라미터 범위가 너무 많을 때 이다. 만약 K-fold 교차검증 방법을 이용한다면 그리드 서치를 수행하는 데 매우 많은 시간이 걸릴 것이다. 그래서 이와 같이 튜닝할 파라미터 범위가 극도로 많을 때는 단순히 1번만 검증을 수행하는 오른쪽의 방법을 이용하는 것이 시간효율적인 측면에서는 유리할 수도 있겠다. 물론 왼쪽 방법을 이용했을 때보다 모델 성능의 객관성은 당연히 낮을 수 밖에 없다는 것은 감안해야 한다.

 

이제 코드로 알아보자. 단지 수행 방법 즉, 사용하는 클래스 이름만 다를 뿐, 스파크의 문법적 차이는 거의 동일하다. 아래 코드를 천천히 살펴보도록 하자.

 

# TrainValidation Split을 활용해서 수행하기
from pyspark.ml.tuning import TrainValidationSplit, ParamGridBuilder

vec_assembler = VectorAssembler(inputCols=train.columns[:-1],
                           outputCol='feature')
train_vec = vec_assembler.transform(train)
test_vec = vec_assembler.transform(test)

dt_clf = DecisionTreeClassifier(featuresCol='feature', labelCol='label')

param_grid = ParamGridBuilder().addGrid(dt_clf.maxDepth, [5, 10]) \
                               .addGrid(dt_clf.minInstancesPerNode, [7, 20]) \
                               .build()
evaluator = MulticlassClassificationEvaluator(labelCol='label',
                                             predictionCol='prediction',
                                             metricName='accuracy')
tvs = TrainValidationSplit(estimator=dt_clf, estimatorParamMaps=param_grid,
                           evaluator=evaluator, trainRatio=0.75, seed=49)

# 학습 데이터
tvs_model = tvs.fit(train_vec)

test_pred = tvs_model.transform(test_vec)
test_eval = MulticlassClassificationEvaluator(labelCol='label',
                                             predictionCol='prediction',
                                             metricName='accuracy')
test_acc = test_eval.evaluate(test_pred)
print('Test Accuracy:', round(test_acc, 3))

3. Pipeline을 활용하기

지금까지 배운 파라미터 튜닝, 교차검증 수행 과정을 하나의 파이프라인으로 활용해서 ML flow 과정을 간단화하여 코드의 만연성을 줄일 수도 있다. Pipeline 클래스를 활용해 파라미터 튜닝, 교차검증 수행하는 과정은 크게 2가지로 사용해 볼 수 있다.

3-1. CrossValidator 안에서 Pipeline을 활용하기

정의한 Pipeline을 CrossValidator의 모델 객체에 넣어주어 활용할 수가 있다. 이에 대해서는 실질적인 코드를 봐야 이해가 수월할 것이다. 먼저 파이프라인을 정의해야 하는데, 여기서는 Feature Vectorization 하는 과정과 모델의 정의하는 부분을 하나의 파이프라인으로 만들었다.

 

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

#====================================
# 1.CrossValidator 안에서 파이프라인을 적용
#====================================
train, test = iris_sdf.randomSplit(weights=[0.7, 0.3], seed=43)
vec_stage = VectorAssembler(inputCols=train.columns[:-1],
                            outputCol='feature')
dt_stage = DecisionTreeClassifier(featuresCol='feature',
                                 labelCol='label')

# [Feature Vectorization -> 모델]을 하나의 파이프라인으로 정의
pipeline_01 = Pipeline(stages=[vec_stage, dt_stage])

 

이제 튜닝할 파라미터 범위와 교차검증을 수행하면서 모델의 성능을 평가할 evaluator를 정의해보자.

 

from pyspark.ml.tuning import ParamGridBuilder
from pyspark.ml.evaluation import MulticlassClassificationEvaluator

# 파라미터 튜닝 grid
param_grid = ParamGridBuilder().addGrid(dt_stage.maxDepth, [8, 20]) \
                               .addGrid(dt_stage.minInstancesPerNode, [5, 10]) \
                               .build()
# 메트릭 evaluator
evaluator = MulticlassClassificationEvaluator(labelCol='label',
                                             predictionCol='prediction',
                                             metricName='accuracy')

 

다음으로는 CrossValidator 클래스를 활용하는데, 이 때 인자에 위에서 정의한 파이프라인 객체를 넣어주도록 하자. 나머지 파라미터 범위, evaluator 를 넣어주는 부분을 동일하다. 그리고 fit() 메소드를 활용해 학습 데이터에 대해 수행하고 반환된 모델 객체의 transform() 메소드로 테스트 데이터에 대해 최종 예측을 수행한다. 이에 대한 코드는 아래와 같다.

 

from pyspark.ml.tuning import CrossValidator

# Cross Validator안에 파이프라인 정의
cv = CrossValidator(estimator=pipeline_01, estimatorParamMaps=param_grid,
                   evaluator=evaluator, numFolds=3)

# 학습 데이터에 대해 파라미터 튜닝 및 교차검증
cv_model = cv.fit(train)

# 테스트 데이터에 대해 최적의 파라미터 모델로 예측 수행
test_pred = cv_model.transform(test)
test_eval = MulticlassClassificationEvaluator(labelCol='label',
                                             predictionCol='prediction',
                                             metricName='accuracy')
test_acc = test_eval.evaluate(test_pred)
print('Test Accuracy:', round(test_acc, 3))

3-2. Pipeline 안에서 CrossValidator를 정의하기

파이프라인을 활용해서 파라미터 튜닝 및 교차검증을 수행하는 또 다른 방법으로는 [3-1.목차] 와는 순서를 반대로 한 방법이다. 즉, 이번에는 Pipeline을 선언할 때 설정하는 stage 중 하나로 CrossValidator를 정의하는 것이다. 이 목차에 대한 코드는 하나씩 설명하지는 않겠다. 단순히 [3-1. 목차]의 ML Flow 순서를 바꾸어 놓은 것이기 때문에 아래 코드의 주석을 읽으면서 이해해보면 그리 어렵지 않을 것이다.

 

from pyspark.ml import Pipeline

#====================================
# 2.파이프라인 안에서 CrossValidator를 적용
#====================================
train, test = iris_sdf.randomSplit(weights=[0.7, 0.3], seed=43)

# 1번째) stage: Feaure Vectorization
vec_stage = VectorAssembler(inputCols=train.columns[:-1],
                           outputCol='feature')
# 2번째) stage: [모델 - param_grid - cv 객체]
dt_clf = DecisionTreeClassifier(featuresCol='feature',
                               labelCol='label')
param_grid = ParamGridBuilder().addGrid(dt_clf.maxDepth, [5, 15]) \
                               .addGrid(dt_clf.minInstancesPerNode, [3, 10]) \
                               .build()
evaluator = MulticlassClassificationEvaluator(labelCol='label',
                                             predictionCol='prediction',
                                             metricName='accuracy')
cv_stage = CrossValidator(estimator=dt_clf,
                         estimatorParamMaps=param_grid,
                         evaluator=evaluator)
# 파이프라인 완성
pipeline_02 = Pipeline(stages=[vec_stage, cv_stage])

# 학습 데이터에 대해 파라미터 튜닝 및 교차검증
pipeline_model = pipeline_02.fit(train)

avg_metrics = pipeline_model.stages[-1].avgMetrics
print('Train Accuracy :',
      round(sum(avg_metrics)/len(avg_metrics), 3)
     )
print()

# 테스트 데이터에 대해 예측 수행
test_pred = pipeline_model.transform(test)
test_eval = MulticlassClassificationEvaluator(labelCol='label',
                                             predictionCol='prediction',
                                             metricName='accuracy')
test_acc = test_eval.evaluate(test_pred)
print('Test Accuracy:', round(test_acc, 3))