머신러닝 프로세스 한 번에 처리하기 | Pipeline, make_pipeline

파이프라인 (Pipeline) 은 전처리의 각 단계, 모델 생성, 학습 등을 포함하는 여러 단계의 머신러닝 프로세스를 한 번에 처리할 수 있는 클래스로, 머신러닝의 sklearn 라이브러리에 내장되어 있습니다.

 

변환기 (Transformer) 

전처리 작업 파이프라인의 경우에는 변환기 (Transformer) 들로만 구성

데이터셋을 기반으로 일련의 모델 파라미터들을 추정

fit_transform()

 

추정기 (Estimator)

전체 프로세스 파이프 라인의 경우에는 여러 개의 변환기와 함께 마지막에 추정기 넣음

데이터셋을 변환하는 추정기

fit()

---

1.

pipe = [ 변환기, 추정기]

pipeline(pipe, verbose = True)​

 

2.

make_pipeline(변환기, 추정기)

 

샘플

import numpy as np
from sklearn.metrics import mean_squared_error
from sklearn import model_selection
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LinearRegression, RidgeCV, LassoCV, ElasticNetCV

class ModelOptimize:

    def __init__(self, X, y, model_name):
    
        self.X = X
        self.y = y
        self.model_name = model_name
        self.X_train, self.X_test, self.y_train, self.y_test = model_selection.train_test_split(self.X, self.y, test_size=0.2)
        self.default_params = get_search_space(self.model_name)
        self.default_cv = 5
        self.default_njobs = 1

        if self.model_name != 'xgboost':
            self.imputer = SimpleImputer(missing_values=np.nan, strategy='median') 
            self.imputer.fit(self.X) 

    def hyper_params_tuning(self, scorer='RMSE'):        
        if scorer == 'RMSE': scorer = 'neg_root_mean_squared_error'
        elif scorer == 'RMSLE': scorer = 'neg_mean_squared_log_error'
        else:
            raise Exception('invalid scorer')

        if self.model_name == 'xgboost':                    
            self.model = xgboost.XGBRegressor(tree_method='gpu_hist')
            self.model_optimize = model_selection.GridSearchCV(estimator=self.model,
                                                            param_grid=self.default_params,
                                                            cv=self.default_cv,
                                                            n_jobs=self.default_njobs,
                                                            scoring=scorer,
                                                            refit=True,
                                                            verbose=0)
            self.model_optimize.fit(self.X_train, self.y_train,
                                    eval_set=[[self.X_test, self.y_test]],
                                    verbose=0)
            self.best_estimator = self.model_optimize.best_estimator_
        
        elif self.model_name == 'ridge':
            self.X_imputed = self.imputer.transform(self.X)
            self.model = RidgeCV(alphas=self.default_params['alphas'], cv=self.default_cv)
            self.model.fit(self.X_imputed, self.y)
            self.best_estimator = Pipeline([('imputer', self.imputer), ('ridge', self.model)])
        elif self.model_name == 'lasso':
            self.X_imputed = self.imputer.transform(self.X)
            self.model = LassoCV(cv=self.default_cv)
            self.model.fit(self.X_imputed, self.y)
            self.best_estimator = Pipeline([('imputer', self.imputer), ('lasso', self.model)])
        elif self.model_name == 'elasticnet':
            self.X_imputed = self.imputer.transform(self.X)
            self.model = ElasticNetCV(l1_ratio=self.default_params['l1_ratio'] , cv=self.default_cv)
            self.model.fit(self.X_imputed, self.y)
            self.best_estimator = Pipeline([('imputer', self.imputer), ('elasticnet', self.model)])
        else:
            raise Exception('invalid model name')
        
    def drop_model(self, file_name, drop_dir='./models'):
        if self.best_estimator:
            os.makedirs(drop_dir, exist_ok=True)
            joblib.dump(self.best_estimator, f'{drop_dir}/{file_name}', compress = 1)
        else:
            raise Exception("HPO not executed yet")
        
    def predict(self, X):
        return self.best_estimator.predict(X)

    def score_rmse(self, X, y):
        return np.sqrt(mean_squared_error(y, self.best_estimator.predict(X)))