JYP Garden

ML 파이프라인 (Machine Learning Pipeline)

Properties1
tagsconcept, mlops, knowledge

6 min read

ML 파이프라인 (Machine Learning Pipeline)

데이터 전처리, 특성 엔지니어링, 모델 훈련을 하나의 재사용 가능한 단위로 통합하는 설계 패턴. 데이터 누수를 방지하고 프로덕션 배포를 간소화한다.

설명

ML 파이프라인은 “재현 가능한 워크플로우”이다. 각 단계가 순서대로 실행되며, 훈련·테스트·프로덕션 환경에서 동일한 변환이 자동으로 적용된다.

구성:

  1. Transformer: 데이터 변환 (스케일링, 인코딩, 결측치 대체)
  2. Estimator: 예측 모델 (분류, 회귀)

핵심 문제: 데이터 누수 (Data Leakage)

상황: 정보 누출

[전체 데이터]
    ↓
[스케일링] ← 평균·표준편차 계산 (전체 데이터 포함!)
    ↓
[Train-Test 분할]
    ├─ 훈련 데이터: 스케일링됨
    └─ 테스트 데이터: 스케일링됨 (훈련 통계 기반)

문제: 테스트 데이터의 통계가 훈련에 영향 → 모델 성능 과평가

해결책: 파이프라인으로 자동화

from sklearn.pipeline import Pipeline
 
pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('classifier', LogisticRegression())
])
 
# 훈련 시: 스케일러가 X_train의 통계 학습
pipe.fit(X_train, y_train)
 
# 테스트 시: 훈련된 스케일러만 적용 (재학습 X)
pipe.predict(X_test)

결과:

  • 스케일러는 X_train에서만 학습
  • X_test는 X_train의 규칙으로만 변환
  • 현실적인 성능 평가 가능

ColumnTransformer: 혼합 자료형 처리

실무 데이터는 수치형(연속변수) + 범주형(카테고리)이 함께 있다.

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
 
ct = ColumnTransformer([
    ('numeric', Pipeline([
        ('imputer', SimpleImputer(strategy='mean')),  # 결측치 → 평균
        ('scaler', StandardScaler())                 # 정규화
    ]), numeric_columns),
    
    ('categorical', Pipeline([
        ('imputer', SimpleImputer(strategy='most_frequent')),  # 결측치 → 최빈값
        ('encoder', OneHotEncoder())                          # 범주 → 0/1
    ]), categorical_columns)
])
 
pipe = Pipeline([
    ('preprocessor', ct),
    ('classifier', LogisticRegression())
])

동작:

  1. 수치형 브랜치: 결측치(평균) → 정규화
  2. 범주형 브랜치: 결측치(최빈값) → 원-핫 인코딩
  3. 두 결과 병렬 실행 후 좌우 연결
  4. 로지스틱 회귀에 입력

하이퍼파라미터 최적화: GridSearchCV

파이프라인의 모든 단계 파라미터를 체계적으로 탐색:

from sklearn.model_selection import GridSearchCV
 
param_grid = {
    'preprocessor__numeric__scaler__with_mean': [True, False],
    'classifier__C': [0.1, 1, 10],
    'classifier__penalty': ['l1', 'l2']
}
 
gs = GridSearchCV(pipe, param_grid, cv=5)
gs.fit(X_train, y_train)
 
print(f"최고 점수: {gs.best_score_:.3f}")
print(f"최적 파라미터: {gs.best_params_}")

명명 규칙:

  • preprocessor: ColumnTransformer 이름
  • numeric: 수치형 브랜치 이름
  • scaler: StandardScaler 이름
  • with_mean: 스케일러 파라미터

4대 이점

1. 데이터 누수 방지

훈련·테스트 데이터 격리 → 현실적 성능 평가

2. 코드 정리 & 유지보수성

전처리와 모델이 분리 → 각각 독립적 수정 가능

# 스케일러만 교체
pipe.named_steps['preprocessor']

3. 교차검증 간편화

from sklearn.model_selection import cross_val_score
 
scores = cross_val_score(pipe, X, y, cv=5)

각 폴드에서 자동으로 전처리 규칙 재학습 → 일관성 보장

4. 모델 배포 & 재사용

import joblib
 
# 저장
joblib.dump(pipe, 'model.pkl')
 
# 프로덕션 로드
pipe = joblib.load('model.pkl')
y_pred = pipe.predict(X_new)

파이프라인 전체(전처리 + 모델)를 단일 파일로 배포 → 재현성 보장

성능 최적화

캐싱 (Caching)

교차검증 중 fit된 변환자 재사용:

pipe = Pipeline([...], memory='/tmp/cache')
cross_val_score(pipe, X, y, cv=5)  # fit된 변환자 캐시 재사용

효과: 계산 시간 50% 단축

병렬화 (Parallelization)

gs = GridSearchCV(pipe, param_grid, n_jobs=-1)

모든 CPU 코어 활용 → 그래프 탐색 속도 대폭 향상

커스텀 변환자

BaseEstimator + TransformerMixin 상속으로 확장:

from sklearn.base import BaseEstimator, TransformerMixin
 
class MyTransformer(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        self.mean_ = X.mean()
        return self
    
    def transform(self, X):
        return X - self.mean_
 
pipe = Pipeline([
    ('custom', MyTransformer()),
    ('classifier', LogisticRegression())
])

주의사항

실수결과해결책
전체 데이터 전처리 후 분할데이터 누수파이프라인으로 자동화
fit_transform을 테스트에 사용테스트 규칙 학습(오염)transform만 사용
파이프라인 외부에서 스케일링일관성 부족모든 변환을 파이프라인 내부에

관련 개념

실전 적용

scikit-learn 의 구현 도구

  • Pipeline, ColumnTransformer, GridSearchCV
  • Transformer 클래스 (StandardScaler, OneHotEncoder)
  • Estimator 클래스 (LogisticRegression, RandomForest)

pandas-dataframe 과의 연관

  • 입력 데이터 소스
  • 혼합 자료형 데이터 처리

소스

그래프 뷰