[SparkML] MLlib Parameter 튜닝 개념 및 코드

 

Parameter Tuning

머신러닝을 다뤄본 사람이라면 익숙한 개념일 것이다.

머신러닝을 알고리즘을 이용하더라도, 그 안의 변수를 어떻게 조정하냐에 따라서 모델의 성능이 달라진다.

MLlib은 당연히도 Paramter Tuning 기능을 제공한다.

코드를 살펴보며 어떻게 튜닝을 하는지 살펴보자.

 

 

 

 

[만약 Spark에 대해 아무것도 모른다면?]

https://mengu.tistory.com/26?category=932924 

[Spark] Apache Spark 개념 및 버전

 

 

 

[SparkML을 모른다면?]

https://mengu.tistory.com/56?category=932924 

[Spark] MLlib 개념 및 실습 코드

 

 

 

[실습 데이터를 가지고 있지 않다면?]

https://mengu.tistory.com/50?category=932924 

[SparkSQL] 택시 데이터 다운/전처리/분석 feat. TLC

 

 

 

"본 포스팅은 패스트캠퍼스의 강의를 듣고, 정리한 자료임을 밝힙니다."

 

 

 

 

 

Basic Settings

기본 세팅이다. 임포트 해줘야 할 것들은 미리 해놨고, SparkSession을 열어두었다.

 

# 폰트 설정
from matplotlib import font_manager, rc
font_path = 'C:\\WINDOWS\\Fonts\\HBATANG.TTF'
font = font_manager.FontProperties(fname=font_path).get_name()
rc('font', family=font)

# basic settings
import os
import findspark
findspark.init(os.environ.get("SPARK_HOME"))
import pyspark
from pyspark import SparkConf, SparkContext
import pandas as pd
import faulthandler
faulthandler.enable()
from pyspark.sql import SparkSession

MAX_MEMORY = "5g"
spark = SparkSession.builder.master('local').appName("taxi-fare-prediction")\
.config("spark.executor.memory", MAX_MEMORY)\
.config("spark.driver.memory", MAX_MEMORY).getOrCreate()

# 데이터가 있는 파일
zone_data = "C:/DE study/data-engineering/01-spark/data/taxi_zone_lookup.csv"
trip_files = "C:/DE study/data-engineering/01-spark/data/trips/*"

# 데이터 로드
trips_df = spark.read.csv(f"file:///{trip_files}", inferSchema = True, header = True)
zone_df = spark.read.csv(f"file:///{zone_data}", inferSchema = True, header = True)

# 데이터 스키마
trips_df.printSchema()
zone_df.printSchema()

# 데이터 createOrReplaceTempView()
trips_df.createOrReplaceTempView("trips")
zone_df.createOrReplaceTempView("zone")


root
 |-- VendorID: integer (nullable = true)
 |-- tpep_pickup_datetime: string (nullable = true)
 |-- tpep_dropoff_datetime: string (nullable = true)
 |-- passenger_count: integer (nullable = true)
 |-- trip_distance: double (nullable = true)
 |-- RatecodeID: integer (nullable = true)
 |-- store_and_fwd_flag: string (nullable = true)
 |-- PULocationID: integer (nullable = true)
 |-- DOLocationID: integer (nullable = true)
 |-- payment_type: integer (nullable = true)
 |-- fare_amount: double (nullable = true)
 |-- extra: double (nullable = true)
 |-- mta_tax: double (nullable = true)
 |-- tip_amount: double (nullable = true)
 |-- tolls_amount: double (nullable = true)
 |-- improvement_surcharge: double (nullable = true)
 |-- total_amount: double (nullable = true)
 |-- congestion_surcharge: double (nullable = true)

root
 |-- LocationID: integer (nullable = true)
 |-- Borough: string (nullable = true)
 |-- Zone: string (nullable = true)
 |-- service_zone: string (nullable = true)

 

 

 

 

 

 

 

DataFrame

---

데이터를 일단 DataFrame 형식으로 가져와야 한다. 또한 원하는 Input과 Output을 남기고는 제거해둬야 한다.

 

 

query = '''
SELECT
    trip_distance,
    total_amount
FROM
    trips
WHERE
    total_amount < 5000
    AND total_amount > 0
    AND trip_distance > 0
    AND trip_distance < 500
    AND passenger_count < 4
    AND TO_DATE(tpep_pickup_datetime) >= '2021-01-01'
    AND TO_DATE(tpep_pickup_datetime) < '2021-08-01'
'''

# 원하는 data만 남겨두기
data_df = spark.sql(query)
data_df.createOrReplaceTempView('data')


data_df.show


+-------------+------------+
|trip_distance|total_amount|
+-------------+------------+
|         16.5|       70.07|
|         1.13|       11.16|
|         2.68|       18.59|
|         12.4|        43.8|
|          9.7|        32.3|
|          9.3|       43.67|
|         9.58|        46.1|
|         16.2|        45.3|
|         3.58|        19.3|
|         0.91|        14.8|
|         2.57|        12.8|
|          0.4|         5.3|
|         3.26|        17.3|
|        13.41|       47.25|
|         18.3|       61.42|
|         1.53|       14.16|
|          2.0|        11.8|
|         16.6|       54.96|
|         15.5|       56.25|
|          1.3|        16.8|
+-------------+------------+
only showing top 20 rows

 

 

 

이번 포스팅의 목적은 파라미터 튜닝이다. 일정한 양의 데이터를 가지고 계속 모델을 테스트하여, 어떤 파라미터가 좋은 성능을 내는지 비교 분석해야 한다. 그렇다면 데이터의 크기는 얼마로 잡아야 할까?

 

 

toy_df = data_df.sample(False, 0.01, seed=1)

 

 

보통은 원래 데이터의 10%를 뽑아서 사용한다. 하지만 기존 데이터의 크기가 너무 큰 경우, 컴퓨터가 터져버릴 수 있다. 택시 데이터는 크기가 어마어마하기에... 그리고 10%를 뽑아서 돌려봤는데 컴퓨터가 터져서 1%로 설정했다.

 

 

 

 

 

 

전처리 & Pipeline

---

이전 포스팅에서 진행했던 전처리 파이프라인을 그대로 만들어 준다.

 

 

 

(1) 원-핫 인코딩

카테고리형 칼럼만 원-핫 인코딩을 해주면 된다. 

 

- StringIndexer : 카테고리형을 원-핫 인코딩하기 전에, 숫자를 먼저 부여해준다. ex) (딸기, 바나나, 초코) -> (1, 3, 2)

- OneHotEncoder : 숫자 부여된 것을 벡터화한다. ex) 1 -> [1, 0, 0]

 

# 원-핫 인코딩
from pyspark.ml.feature import OneHotEncoder, StringIndexer


# 카테고리형 칼럼
cat_feats = [
    'pickup_location_id',
    'dropoff_location_id',
    'day_of_week'
]

# 파이프라인 구성을 위한 stages 리스트
stages = []


# index를 바꾸고, 그 바뀐 indexer에 원-핫 인코딩을 적용해준다.
for c in cat_feats:
    cat_indexer = StringIndexer(inputCol=c, outputCol = c + "_idx").setHandleInvalid("keep")
    onehot_encoder = OneHotEncoder(inputCols = [cat_indexer.getOutputCol()], outputCols=[c + '_onehot'])
    stages += [cat_indexer, onehot_encoder]

 

 

 

 

(2) 숫자형 데이터, 벡터화 및 스케일러 적용

숫자형 데이터를 하나의 벡터로 묶고, 스케일러를 적용한다.

 

# 벡터화 + 스케일러 적용
from pyspark.ml.feature import VectorAssembler, StandardScaler

# numeric형 칼럼
num_feats = [
    'passenger_count',
    'trip_distance',
    'pickup_time'
]


# 벡터화 한 후, 스케일러를 적용한다.
for n in num_feats:
    num_assembler = VectorAssembler(inputCols=[n], outputCol= n + '_vecotr')
    num_scaler = StandardScaler(inputCol=num_assembler.getOutputCol(), outputCol = n + '_scaled')
    stages += [num_assembler, num_scaler]

 

 

* 중간 stages 점검

 

stages


[StringIndexer_115f74e6efea,
 OneHotEncoder_714f494271bb,
 StringIndexer_806a3b8e8a32,
 OneHotEncoder_f7bc9266f650,
 StringIndexer_2f125ebb95a8,
 OneHotEncoder_eb212d50e427,
 VectorAssembler_5b7bfff3be42,
 StandardScaler_3e59d49af9ad,
 VectorAssembler_a61858dada0f,
 StandardScaler_c739fc7f7d49,
 VectorAssembler_f683b2eeb4d2,
 StandardScaler_96e5ba925088]

 

 

 

 

(3) 모두  VectorAssemble 하여 훈련에 적합한 데이터 셋을 완성하도록 구축하자.

 

# inputs 칼럼
assembler_inputs = [c + '_onehot' for c in cat_feats] + [n + '_scaled' for n in num_feats]
print(assembler_inputs)


['pickup_location_id_onehot',
 'dropoff_location_id_onehot',
 'day_of_week_onehot',
 'passenger_count_scaled',
 'trip_distance_scaled',
 'pickup_time_scaled']
 
 
 
 # stages에 VectorAssemble 추가
assembler = VectorAssembler(inputCols=assembler_inputs, outputCol= 'feature_vector')
stages += [assembler]
stages


[StringIndexer_dfc09cc586be,
 OneHotEncoder_987fbfa36a2d,
 StringIndexer_bf2338365d7f,
 OneHotEncoder_5a91ea5195e8,
 StringIndexer_c416d64272f1,
 OneHotEncoder_0dfab0742066,
 VectorAssembler_4c5f47a3740c,
 StandardScaler_65dfe2363318,
 VectorAssembler_7e0a4e81ec39,
 StandardScaler_22d11d283c0a,
 VectorAssembler_c2b692153924,
 StandardScaler_debc924ffa61,
 VectorAssembler_c2c382815ebb]

 

 

 

 

(4) Final, Pipeline()을 통해서 다 묶어주기

 

# pipeline 만들기
from pyspark.ml import Pipeline

transform_stages = stages
pipeline = Pipeline(stages = transform_stages)
fitted_transformer = pipeline.fit(train_df)

 

 

 

 

 

Hyperparameter Tuning

---

 

from pyspark.ml import Pipeline
from pyspark.ml.regression import LinearRegression
from pyspark.ml.tuning import CrossValidator, ParamGridBuilder
from pyspark.ml.evaluation import RegressionEvaluator

 

 

(1) 파이프라인에 선형 회귀 추가하기

 

lr = LinearRegression(
    maxIter=30,
    solver='normal',
    labelCol='total_amount',
    featuresCol='feature_vector'
)

cv_stages = stages + [lr]


# 파이프라인 완성하기
cv_pipeline = Pipeline(stages = cv_stages)

 

 

 

(2) Parameter map 작성하기

MLlib에선 Parameter map을 따로 작성한 후, 교차 검증의 변수로 집어넣어야 한다. 그래야 Parameter 들을 순환하여 모든 경우의 수를 돌아볼 수 있다.

 

lr.elasticNetParam -> 선형 회귀의 elasticNet의 alpha 파라미터를 말한다.

lr.regParam -> 선형 회귀의 regParam 파라미터를 의미한다.

 

# '\' 는 줄바꿈을 가능하게 하는 문자로 쓰인다.
param_grid = ParamGridBuilder()\
.addGrid(lr.elasticNetParam, [0.1, 0.2, 0.3, 0.4, 0.5])\
.addGrid(lr.regParam, [0.01, 0.02, 0.03, 0.04, 0.05]).build()

 

 

 

(3) 교차 검증 함수 만들기

 

# estimator에는 파이프라인을,
# ParamMaps에는 파라미터 맵을
# evaluator에는 평가 지표를,
# numFolds에는 교차 검증의 폴드 수를 입력했다.

cross_val = CrossValidator(estimator = cv_pipeline,
                           estimatorParamMaps=param_grid,
                           evaluator=RegressionEvaluator(labelCol='total_amount'),
                           numFolds=5)

 

 

 

(4) 교차 검증 Start

 

cv_model = cross_val.fit(toy_df)

 

 

 

(5) Best Parameter 뽑아내기

 

alpha = cv_model.bestModel.stages[-1]._java_obj.getElasticNetParam()
reg_param = cv_model.bestModel.stages[-1]._java_obj.getRegParam()

print(f'alpha is {alpha}')
print(f'reg_param is {reg_param}')


alpha is 0.4
reg_param is 0.05

 

 

 

 

 

 

Training & 추론

---

 

(1) train/test 데이터를 파이프라인 통과시키기

 

transform_stages = stages
pipeline = Pipeline(stages = transform_stages)
fitted_transformer = pipeline.fit(train_df)


vtrain_df = fitted_transformer.transform(train_df)
vtest_df = fitted_transformer.transform(test_df)

 

 

 

(2) Training

 

lr = LinearRegression(
    maxIter=50,
    solver='normal',
    labelCol='total_amount',
    featuresCol='feature_vector'
    elasticNetParam = alpha,
    regParam = reg_param
)


model = lr.fit(vtrain_df)

 

 

 

(3) 추론 및 성능 평가

 

prediction = model.transform(vtest_df)


predictions.select(["trip_distance", "day_of_week", "total_amount", "prediction"]).show()


+-------------+-----------+------------+------------------+
|trip_distance|day_of_week|total_amount|        prediction|
+-------------+-----------+------------+------------------+
|          3.1|   Saturday|       22.55| 18.44882896087039|
|          7.9|   Saturday|        30.3|28.898380309569866|
|          1.4|   Saturday|        16.0|13.635679102878225|
|          1.1|    Tuesday|       12.95|14.050765065622219|
|          0.7|   Saturday|         9.8|12.151950922741019|
|          1.1|     Monday|        11.8|14.302981883348586|
|          3.7|     Friday|       24.35|19.745504411653762|
|          2.4|   Saturday|       14.75|16.012957291356248|
|          3.4|   Saturday|       20.15|17.933430832644525|
|          6.5|     Friday|       32.75|27.541169751290077|
|          2.6|   Saturday|        17.8|16.614148948305857|
|          0.2|   Saturday|        5.15| 7.565190314421683|
|          0.9|     Sunday|         6.3| 9.491154409419867|
|          9.9|   Saturday|       46.35|38.455267206674144|
|          4.5|   Saturday|       22.85|15.766824456415733|
|          0.6|    Tuesday|         4.8| 7.422348054207551|
|          1.4|    Tuesday|        10.8|12.006766375046109|
|          4.3|   Saturday|        20.3|19.040664139319205|
|          3.9|   Thursday|        17.8|18.389827313235557|
|          5.8|    Tuesday|        27.0|23.073838896480723|
+-------------+-----------+------------+------------------+
only showing top 20 rows

 

 

 

 

RMSE

 

# before
model.summary.rootMeanSquaredError

5.818945295076586



# After
model.summary.rootMeanSquaredError

5.610492491836879

 

 

 

 

 

R2

 

# before
model.summary.r2

0.7997047915616821



# After
model.summary.r2

0.8108436137289087

 

 

파라미뉴 튜닝을 하기 전과 성능 차이가 꽤 난다는 것을 확인할 수 있다.

 

 

 

 

 

 

모델 저장 및 불러오기

---

# 저장하기
model_dir = "저장 경로"
model.save(model_dir)


# 불러오기
from pyspark.ml.regression import LinearRegressionModel
lr_model = LinearRegressionModel().load(model_dir)


# 바로 추론하기
predictions = lr_model.transform(vtest_df)

파라미터를 튜닝하고, 최적의 파라미터로 모델의 성능을 높여보았다. 더 나아가 모델을 저장하고, 어떻게 불러오는지도 확인했다. 이제 스스로 다른 모델도 적용하면서 MLlib에 익숙해져 보자!

GOOD BYE!