[PySpark] PySpark로 Regression 모델 만들기

🔊 본 포스팅은 PySpark를 활용한 Kaggle Notebook을 필사하면서 배우게 된 코드 내용을 기반으로 포스팅했음을 알려드립니다. 또한 앞으로 소개될 PySpark의 문법에 대해서 생소하시다면 여기를 참고해 간단한 예시를 통해 이해를 할 수 있습니다.

 

이번 포스팅에서는 PySpark 그 중에서도 Spark SQL 과 Spark MLlib을 활용한 머신러닝 회귀 모델을 만드는 방법에 대해 소개하려고 한다. 활용한 데이터는 California Housing 데이터를 사용했다. 데이터는 여기를 눌러 하단의 Input 목차를 보면 다운로드 받을 수 있다.

 

PySpark로 주택 가격을 예측하는 Regression 모델을 만들어보자!

 

이전 포스팅에서 실습해봤던 분류 모델과 큰 프레임이 다르진 않다. 따라서 이번 포스팅에서는 수치형 변수에 주로 적용하는 기술통계량 값을 보는 방법이나 Scaling 방법을 PySpark로 어떻게 하는지에 대해 초점을 맞추어 실습해 나가면 좋을 것 같다.

1. 필요한 라이브러리와 데이터 로드

항상 그래왔듯이 분석 및 모델링을 하기 위해 필요한 라이브러리들과 데이터를 로드해보자. 먼저 필요한 라이브러리들이다.

 

import os
import pandas as pd
import numpy as np

# pyspark for objecy, sql
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession, SQLContext

from pyspark.sql.types import *
import pyspark.sql.functions as F
from pyspark.sql.functions import udf, col

# pyspark for ML
from pyspark.ml.regression import LinearRegression
from pyspark.mllib.evaluation import RegressionMetrics
from pyspark.ml.evaluation import RegressionEvaluator

from pyspark.ml.tuning import ParamGridBuilder, CrossValidator, CrossValidatorModel
from pyspark.ml.feature import VectorAssembler, StandardScaler

import seaborn as sns
import matplotlib.pyplot as plt

# Setting for visualization
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = 'all'

pd.set_option('display.max_columns', 200)
pd.set_option('display.max_colwidth', 400)

from matplotlib import rcParams
sns.set(context='notebook', style='whitegrid',
       rc={'figure.figsize': (18, 4)})
rcParams['figure.figsize'] = 18, 4

# 노트북 재실행을 대비하기 위해 랜덤 시드 설정해놓기
rnd_seed = 23
np.random.seed = rnd_seed
np.random.set_state = rnd_seed

 

그런데 이번 데이터는 특이하게 데이터가 data 확장명으로 된 파일이며 데이터의 header 즉, 칼럼 네임들이 없기 때문에 사용자가 직접 스키마를 정의해주고 프로그래밍으로 명시해주어야 한다. 이에 대한 부분은 이전 포스팅 4번 목차에서 다루었으니 참고 바란다.

 

## 데이터 스미카 직접 명시해주기 ##
data_path = '/Users/younghun/Desktop/gitrepo/data/cal_housing.data'
# data의 칼럼명을 스키마로 정의해주기
schema_string = 'long lat medage totrooms totbdrms pop houshlds medinc medhv'

fields = [StructField(field, FloatType(), True)for field in schema_string.split()]
schema = StructType(fields)

## 데이터 로드를 위한 SparkSession 만들어주기 ##
spark = SparkSession.builder.master('local[2]')\
        .appName('Linear-Regression-California-Housing')\
        .getOrCreate()

# 데이터파일 로드
# cache 메소드를 이용해서 메모리에 keep해놓기
housing_df = spark.read.csv(path=data_path, schema=schema).cache()

# 상위 5개 행 미리 보기 -> 하나의 Row가 namedTuple 형태로 되어 있음..!
housing_df.take(5)

 

위 코드에서 특이한 부분은 take()라는 부분을 사용한 것인데, 이전에 사용했던 limit과 show와는 달리 데이터 미리보기를 Python named tuple 형태인 DataFrame의 Row 형식으로 데이터를 보여준다.

 

take는 Row 타입으로 데이터를 미리보여준다.

 

반면에 show 메소드는 다음과 같이 보여준다.

 

show는 데이터프레임 형태로 데이터를 미리보여준다.

2. 데이터 탐색

이전 실습에서 이미 사용해보았던 메소드들은 설명을 생략하고 새롭게 등장하는 메소드에 대해서만 설명하려고 한다.

 

• df.sort('columnA', ascending=True) : columnA 값을 기준으로 오름차순으로 정렬한다. 디폴트값은 True로 False면 내림차순으로 정렬한다. 이는 Pandas API의 df.sort_values(by='columnA', ascending=True) 와 비슷하다.
• df.describe() : 수치형 변수들의 기술통계량을 보여주는 메소드로 Pandas API 형태와 매우 비슷하다. 단, PySpark에서는 기술통계량 값을 직접 관찰하고 싶다면 show()와 같은 미리보기 메소드를 붙여 df.describe().show(10) 이런 식으로 해주어야 한다.

 

# 그룹핑해서 집계해보기
result_df = housing_df.groupBy('medage').count()
# 값 정렬 기준 설정해주고 내림차순으로 정렬
result_df.sort('medage', ascending=False).show(5)

# 수치형 변수들 기술통계량 살펴보기
housing_df.describe().show(10)

 

다음은 수치형 변수 값들의 자릿수를 줄여주는 것과 같이 SQL에서 사용할 수 있는 여러가지 함수가 담겨있는 라이브러리를 사용해보자. 하단의 코드에서 F는 pyspark.sql.functions를 F로 alias 시킨 것이다.

 

# 반환되는 통계량 값들을 소수점 제거하거나 하는 등 커스터마이징해서 출력
(housing_df.describe()).select('summary',
                              F.round('medage', 4).alias('medage'),
                              F.round('totrooms', 4).alias('totrooms'),
                              F.round('totbdrms', 4).alias('totbdrms'),
                              F.round('pop', 4).alias('pop'),
                              F.round('houshlds', 4).alias('houshlds'),
                              F.round('medinc', 4).alias('medinc'),
                              F.round('medhv', 4).alias('medhv')).show(10)

 

round 함수를 활용해 깔끔하게 수치값을 정리해주었다.

3. Feature Engineering

다음은 기존 변수에서 파생변수를 생성하는 작업인데, 이전에 했던 것처럼 withColumn 구문을 사용하여 새로운 파생변수를 생성해주면 된다.

 

# 파생변수 생성하기
housing_df = (housing_df.withColumn('rmsperhh',
                                   F.round(col('totrooms')/col('houshlds'), 2))\
                        .withColumn('popperhh',
                                   F.round(col('pop')/col('houshlds'), 2))\
                        .withColumn('bdrmsperrm',
                                   F.round(col('totbdrms')/col('totrooms'), 2)))
housing_df.show(5)

 

새로운 파생변수 생성하기

 

이제 필요한 변수들만 골라서 데이터프레임을 새로 할당한 후 수치형 변수들에 Scaling을 적용시켜보자. 특이한 점은 Scaling을 적용시키기 이전에 필요한 Feature들을 Vector로 변환하는 작업이 선행되어야 한다는 것이다.

 

# 사용하지 않을 변수 제외하고 필요한 변수들만 select
housing_df = housing_df.select('medhv',
                              'totbdrms',
                              'pop',
                              'houshlds',
                              'medinc',
                              'rmsperhh',
                              'popperhh',
                              'bdrmsperrm')

 featureCols = ['totbdrms', 'pop', 'houshlds', 'medinc',
              'rmsperhh', 'popperhh', 'bdrmsperrm']

# VectorAssembler로 feature vector로 변환
assembler = VectorAssembler(inputCols=featureCols, outputCol='features')
assembled_df = assembler.transform(housing_df)
assembled_df.show(10, truncate=True)

 

VectorAssembler로 수치형 Feature들을 Vector로 변환했다.

 

이제 위 사진의 빨간색 박스안에 있는 수치형 Feature들을 Scaling 시켜보자.

 

# 위에서 만든 Feature vector인 'features' 넣기
standardScaler = StandardScaler(inputCol='features',
                               outputCol='features_scaled')
# fit, transform
scaled_df = standardScaler.fit(assembled_df).transform(assembled_df)

# scaling된 피처들만 추출해보기
scaled_df.select('features', 'features_scaled').show(10,
                                                    truncate=True)

 

Scaling 하는 API도 PySpark에서 제공하는데 여러가지 스케일링 방법이 있지만 여기서는 표준화인 StandardScaler를 사용해보자. 메소드 사용 방법은 다음과 같다. StandardScaler(inputCol='Raw Features', outputCol='Scaled Features') 로 스케일링을 정의해주고 fit(dataframe)과 transform(dataframe)으로 스케일링을 수행해주자.

 

원본 Feature와 스케일링된 Feature

4. 데이터 분할과 회귀 모델링

데이터를 학습 8, 테스트 2의 비율로 나누어주고 회귀 모델을 만들어 학습시켜보자. 회귀 모델은 여러가지가 있지만 여기서는 오버피팅을 예방하기 위한 정규화 항이 포함되어 있는 Elastic Net 회귀 모델을 사용해보자.

 

# 데이터 분할
train_data, test_data = scaled_df.randomSplit([0.8, 0.2], seed=rnd_seed)

# Elastic Net 모델 정의
lr = LinearRegression(featuresCol='features_scaled',
                     labelCol='medhv',
                     predictionCol='predmedhv',
                     maxIter=10,
                     regParam=0.3,
                     elasticNetParam=0.8,
                     standardization=False)
# 모델 학습
linearModel = lr.fit(train_data)

5. 테스트 데이터에 대한 예측

이제 학습을 모두 마쳤으니 테스트 데이터로 예측을 하고 RMSE, MAE, R2 Score 메트릭들로 성능 평가를 해보자. PySpark에서 테스트 데이터에 대한 평가는 transform(test_dataframe)으로 수행해준다.

 

# transoform 사용하면 모델 정의할 때 설정한 "예측 값 변수"를 새로 만들어 생성한 데이터프레임 반환
predictions = linearModel.transform(test_data)

print(type(predictions)) # 예측 결과값이 PySpark의 데이터프레임 형태로 반환됨!

 

변수 prediction이 데이터프레임 형태로 반환되는데 이 데이터프레임 중에 예측값인 predmedhv와 정답인 medhv 칼럼만을 추출해보자. 결과화면을 보면 머신러닝 모델의 예측 성능이 그리 높아보이진 않는다..

 

# predictions 데이터프레임에서 y값과 예측값만 추출해 비교
pred_labels = predictions.select('predmedhv', 'medhv')
pred_labels.show()

 

예측값과 실제값 비교

 

그렇다면 예측값과 실제값이 얼마나 차이나는지 RMSE, MAE를 계산해보고 만들어진 회귀식이 주어진 데이터를 얼마나 잘 설명하는지 나타내는 R2 Score(결정계수)값을 살펴보자.

 

print(f"RMSE:{linearModel.summary.rootMeanSquaredError}")
print(f"MAE: {linearModel.summary.meanAbsoluteError}")
print(f"R2 score: {linearModel.summary.r2}")

 

예측 성능이 그리 높지 않다.

 

추가적으로 메트릭을 얻을 수 있는 방법은 위 코드 말고도 RegressionEvaluator로도 계산할 수 있다. 해당 메소드의 사용방법은 다음과 같다.

 

• RegressionEvaluator(predictionCol, labelCol, metricName)
  • predictionCol : 예측값
  • labelCol : 실제값
  • metricName : 측정할 메트릭 이름(ex. RMSE, MAE 등등)

 

# RMSE
evaluator = RegressionEvaluator(predictionCol='predmedhv',
                               labelCol='medhv',
                               metricName='rmse')
# MAE                            
evaluator = RegressionEvaluator(predictionCol='predmedhv',
                               labelCol='medhv',
                               metricName='mae')
# R2 Score
evaluator = RegressionEvaluator(predictionCol='predmedhv',
                               labelCol='medhv',
                               metricName='r2')

print(f"RMSE: {evaluator.evaluate(pred_labels)}")
print(f"MAE: {evaluator.evaluate(pred_labels)}")  
print(f"R2 score: {evaluator.evaluate(pred_labels)}")

 

또 다른 방법은 RegressionMetrics 메소드인데 이는 특이하게 예측값을 RDD 자료구조로 변환하고 넣어주어야 한다.

 

# 예측값인 pred_labels를 RDD 자료구조로 변환하고 넣어야 함
metrics = RegressionMetrics(pred_labels.rdd)

print("RMSE:", metrics.rootMeanSquaredError)
print("MAE:", metrics.meanAbsoluteError)
print("R2 score:", metrics.r2)