오늘도 꾸준히
ML/DL 학습모델 저장하기((Pickle, dill)) 본문
흔히들 머신러닝이나 딥러닝 모델을 돌리게 되면, 모델의 구조나 데이터 크기에 따라 시간이 기하급수적으로 늘어난다. 종종 2~3일 내내 돌아가는 딥러닝 모델도 존재한다. 학습 모델을 저장하는 이유는 여러 가지가 존재할 수 있는데, 학습 모델을 실제 산업 현장에 적용 한다거나, 크기가 매우 큰 딥러닝 모델을 여러번 나누어 학습할 때 사용할 수 있다.
난 그 중에 학습 모델을 실제 현장에 적용하기 위해 학습 모델을 저장해보고자 한다. 매 번 현장에서 학습 모델을 돌리고 예측값을 생성하면 너무 시간이 오래 걸리기 때문에, 학습된 모델의 가중치를 저장한 뒤, 실제 현장에서는 예측값만 생성하는 것이다. 학습 모델을 저장하는 데에는 Pickle 을 많이 사용하는데, Pickle의 한계점은 무엇이고, dill은 또 무엇인 지 간단하게 알아보고자 한다.
Pickle과 dill은 사실상 같은 역할을 한다. 쉽게 설명하면 모델을 "저장"한다는 의미로 이해할 수 있는데, 정확한 의미는 저장이 아니라 "직렬화((serializing)) 하는 수단"이다. 그렇다면 직렬화란 무엇일까?
직렬화((serializing))
직렬화는 일정한 규칙에 따라 데이터를 byte stream으로 변형하는 것을 의미하는데, 아래 블로그를 참고하면 쉽게 이해할 수 있다.
* 직렬화 https://itholic.github.io/python-serialize/
[python] serialize (직렬화)
직렬화
itholic.github.io
해당 블로그에선 햄버거를 예시로 설명하는데, 햄버거에는 빵, 양상추, 패티, 치즈 등 여러 요소가 존재한다. 우리는 이 햄버거를 특정 컨베이어 벨트를 통해 이동시키려 하는데, 컨베이어 벨트의 크기가 매우 작다. 한 번에 딱 한가지 요소만 들어갈 정도로 작은데, 우리는 이 때 햄버거를 각각의 요소별로 해체한 뒤 이동시킬 수 있다. 그리고 햄버거가 다 이동하면, 원래의 규칙을 역으로 하여 기존의 햄버거로 다시 조립한다. 직렬화는 이와 비슷하다. 복잡하게 짜여진 데이터를 요소별로 나열한 뒤 저장한다고 이해할 수 있다. 그렇다면, Pickle과 dill은 어떤 차이가 있을까?
Pickle과 dill의 가장 큰 차이는 활용할 수 있는 데이터 형태의 범위이다. 이게 무슨 뜻이냐면, Pickle은 직렬화가 가능한 데이터 형태에만 적용할 수 있다. 하지만, dill은 직렬화가 되지 않는 데이터 형태에도 적용할 수 있다. 예를 들어 lambda, class 등이 dill로 적용할 수 있는 형태이다. 필자는 class 자체를 저장하기 위해 dill을 사용하였고, dill이 제공하는 종류는 매우 다양하다.
하지만 여전히 dill을 사용하더라도 저장할 수 없는 형태가 존재한다.
Pickle과 dill에 관한 정보는 아래 블로그에서 참고 하였으며, 더 자세한 내용은 공식 문서를 확인하길 바란다.
* Python dill로 class definition까지 binary로 저장하기( https://lovit.github.io/analytics/2019/01/15/python_dill/ )
Python dill 로 class definition 까지 binary 로 저장하기
파이썬으로 작업을 할 때, 사용자가 정의하는 클래스 인스턴스를 저장할 일들이 있습니다. 예를 들면 namedtuple 을 이용한 데이터 타입이라던지, PyTorch 에서 nn.Module 을 상속받은 모델들이 그 예입
lovit.github.io
이제, dill을 사용하여 학습 모델을 저장하고 이를 현장에서 사용할 수 있는 형식으로 만들어보고자 한다.
우선 모델을 학습할 Class를 만들어줘야 한다. 필자는 한 클래스 안에 3개의 모델이 한 번에 돌아가도록 구상하였다.
import pandas as pd
import numpy as np
import dill
from sklearn.linear_model import LinearRegression
from keras.models import Sequential
from keras.layers import Conv1D, MaxPooling1D, Dense, Dropout, GlobalAveragePooling1D, BatchNormalization, Activation
from keras.callbacks import EarlyStopping
from tensorflow.keras.optimizers import Adam
class Predictor:
def __init__(self, train):
self.train = train.bfill()
self.linear_model = None
self.residual_trend_model = None
self.residual_model = None
def split_data(self): # x,y 분리
x_cols = self.train.drop(columns=['y']).columns
train_x = self.train[x_cols]
train_y = self.train['y']
return train_x, train_y
# 선형 모델
def inear_model(self):
train_x, train_y = self.split_data()
self.linear_model = LinearRegression()
self.linear_model.fit(train_x, train_y)
pred_train_y = self.linear_model.predict(train_x)
# 잔차 생성
residual = train_y - pred_train_y # 실제값 - 예측값으로 잔차 계산
return pred_train_y, residual
# 잔차 전처리(시차변수 생성) 생략
# 잔차 트렌드 예측 모델
def train_residual_trend_model(self):
train_x_res, train_y_res = self.residual_preprocessing()
train_x_res_trend = train_x_res.copy()
train_y_res_trend = train_y_res.rolling(window=24).mean().bfill() # 타겟변수를 트렌드로 변경
# 데이터 3차원으로 변환 -> 1D CNN -> (batch, timestep, feature)
train_x_res_trend = train_x_res_trend.values.reshape(len(train_x_res_trend), 1, len(train_x_res_trend.columns))
# 모델 정의
self.trend_model = Sequential([
Conv1D(128, 3, padding='same', input_shape=(train_x_res_trend.shape[1], train_x_res_trend.shape[2])),
BatchNormalization(),
Activation('relu'),
MaxPooling1D(2, padding='same'),
Conv1D(64, 3, padding='same'),
BatchNormalization(),
Activation('relu'),
MaxPooling1D(2, padding='same'),
GlobalAveragePooling1D(),
Dense(64, activation='relu'),
Dense(32, activation='relu'),
Dense(16, activation='relu'),
Dropout(0.3),
Dense(1)
])
self.trend_model.compile(loss='mse', optimizer=Adam(learning_rate=0.000005), metrics=['mae', 'mse'])
early_stop = EarlyStopping(monitor='val_loss', patience=10)
history = self.trend_model.fit(train_x_res_trend, train_y_res_trend,
epochs=50,
batch_size=64,
callbacks=[early_stop])
pred_train_resid = self.trend_model.predict(train_x_res_trend)
return pred_train_resid
# 잔차 예측 모델(예측된 트렌드 사용)
def residual_prediction_model(self):
train_x_res, train_y_res = self.residual_preprocessing()
# 트렌드 예측값 독립변수로 추가
train_x_res['trend_pred'] = self.trend_model.predict(train_x_res.values.reshape(len(train_x_res), 1, len(train_x_res.columns)))
# 데이터 3차원으로 변환 -> 1D CNN -> (batch, timestep, feature)
train_x_res = train_x_res.values.reshape(len(train_x_res), 1, len(train_x_res.columns))
# 모델 정의
self.residual_model = Sequential([
Conv1D(128, 3, padding='same', input_shape=(train_x_res.shape[1], train_x_res.shape[2])),
BatchNormalization(),
Activation('relu'),
MaxPooling1D(2, padding='same'),
Conv1D(64, 3, padding='same'),
BatchNormalization(),
Activation('relu'),
MaxPooling1D(2, padding='same'),
Conv1D(64, 3, padding='same'),
BatchNormalization(),
Activation('relu'),
MaxPooling1D(2, padding='same'),
GlobalAveragePooling1D(),
Dense(64, activation='relu'),
Dense(32, activation='relu'),
Dense(16, activation='relu'),
Dense(8, activation='relu'),
Dropout(0.3),
Dense(1)
])
self.residual_model.compile(loss='mse', optimizer=Adam(learning_rate=0.00001), metrics=['mae', 'mse'])
early_stop = EarlyStopping(monitor='val_loss', patience=10)
history = self.residual_model.fit(train_x_res, train_y_res, epochs=40, batch_size=64, callbacks=[early_stop])
pred_train_resid = self.residual_model.predict(train_x_res)
return pred_train_resid
# 모델 전체 학습
def train_all(self):
self.linear_model()
self.train_residual_trend_model()
self.residual_prediction_model()
# 모델 저장
def save(self, filepath):
with open(filepath, 'wb') as f:
dill.dump(self, f) # dill로 모델 저장
# 예측 함수(test에서 사용)
def predict(self, test):
x_cols = test.drop(columns=['y']).columns
x_x = test[x_cols]
x_y = test['y']
linear_pred = self.linear_model.predict(x_x) # 선형 모델
# 잔차 데이터 준비
x_res = x_x.copy()
residual = x_y - linear_pred # 실제값 - 예측값으로 잔차 계산
# 트렌드 예측
x_res_trend = x_res.copy()
x_res_trend = x_res_trend.values.reshape(len(x_res_trend), 1, len(x_res_trend.columns))
trend_pred = self.trend_model.predict(x_res_trend)
# 잔차 예측
x_res['trend_pred'] = trend_pred
x_res = x_res.values.reshape(len(x_res), 1, len(x_res.columns))
residual_pred = self.residual_model.predict(x_res)
# 최종 예측
return linear_pred + residual_pred.flatten()
if __name__ == '__main__':
train = pd.read_csv('train.csv')
predictor = Predictor(train)
predictor.train_all()
predictor.save('predictor.pkl')
해당 클래스는 선형 예측 -> 잔차 트렌드 예측 -> 잔차 예측 순으로 이어지고, 최종적으로 선형 예측값과 잔차 예측값을 더해서 오차를 보정하는 방식이다. 위 코드에서 주의깊게 봐야할 함수는 "save"와 "predict"이다. save는 dill을 사용하여 학습된 모델을 저장하는 코드이고, predict 코드는 예측 생성 파일((ex. test.py))에서 사용할 코드이다.
마지막 부분에서 predictor라는 변수에 전체 클래스를 저장하는 코드가 존재하는데, class 자체를 저장하는 방식이다. 따라서, 예측 생성 파일에서는 dill을 불러온 뒤, predict 함수를 사용할 수 있다.
import pandas as pd
import dill
# 모델 로드
with open('predictor.pkl', 'rb') as f:
model = dill.load(f)
# 테스트 데이터 로드
test = pd.read_csv('test.csv').bfill()
# 예측
predictions = model.predict(test)
# 결과 저장
pd.DataFrame(predictions, columns=['prediction']).to_csv('predictions.csv', index=False)
이제 예측 생성 파일에서 저장된 pkl 파일을 불러오고, ((dill을 사용해도 pickle 형식으로 저장된다)) model.predict((test))를 사용하니 예측값이 생성되는 것을 확인할 수 있다.
'DeepLearning' 카테고리의 다른 글
| Pyannote 기반 화자 분리((Speech\ diarization)) (0) | 2024.07.28 |
|---|---|
| CNN((Convolutional \ Neural \ Network))로 이미지 분석하기 (1) | 2024.07.23 |
| Gradient Descent 으로 손실함수 최적화 (0) | 2024.07.07 |
