Pandas·NumPy·Scikit-learn으로 데이터 처리 파이프라인 구성하기

집계부터 학습까지 끊기지 않는 처리 흐름

데이터 수집, 전처리, 집계, 모델 학습이 서로 다른 방식으로 관리되면 재실행과 검증이 어려워진다. Pandas의 GroupBy·피벗, NumPy의 벡터화, Scikit-learn의 Pipeline을 조합하면 이 과정을 하나의 반복 가능한 흐름으로 묶을 수 있다.

Pandas의 그룹화는 키 컬럼을 기준으로 데이터를 묶은 뒤 집계나 변환을 수행하는 방식이다. 피벗과 피벗테이블은 긴 형태의 데이터를 넓은 형태로 바꾸어 교차 집계 결과를 만든다.

NumPy 벡터화는 요소별 반복을 파이썬 루프가 아니라 C 수준의 배열 연산으로 처리한다. 브로드캐스팅, 불리언 인덱싱, ufunc을 활용하면 대규모 산술·논리 연산을 다룰 수 있다.

Scikit-learn Pipeline은 전처리와 모델을 순서대로 연결한다. 데이터 누수를 막고 재현성을 확보하는 데 쓰이며, ColumnTransformer, FeatureUnion, 캐시·서치(그리드/랜덤)와 결합해 운영 자동화에 활용할 수 있다.

집계 규칙과 배열 연산을 분리해 다루기

GroupBy-agg-transform 흐름에서는 다중 집계, 사용자 정의 함수, 누락값 처리 규칙을 일관되게 관리해야 한다. pivot_table은 다중 인덱스와 다중 컬럼의 교차 집계를 만들 수 있고, fill_value와 margins를 활용하면 보고서 생성을 자동화할 수 있다.

파생 변수 생성과 데이터 클렌징에는 np.where, ufunc, 불리언 마스크를 사용할 수 있다. 파이썬 for 루프와 비교해 NumPy 벡터화는 10~100배 성능 향상을 기대할 수 있지만, 메모리 레이아웃의 연속성과 dtype 최적화가 결과에 큰 영향을 준다.

학습 단계에서는 숫자·범주·텍스트처럼 성격이 다른 스키마를 ColumnTransformer로 병행 처리한다. Pipeline을 cross_val_score와 함께 사용하면 데이터 누수를 차단할 수 있으며, set_output(transform="pandas")는 전처리 뒤에도 특성명을 추적하는 데 도움이 된다.

검증과 운영 관찰을 처리 단계에 포함하기

입력 단계에서는 열 존재 여부, dtype, 값 범위를 먼저 확인한다. 검증에 실패했을 때는 로깅, 중단, 대체값 전략 중 어떤 처리를 적용할지 명확히 정해야 한다.

집계 테이블, 특성 중요도, 메트릭처럼 단계별 산출물을 저장해 두면 디버깅과 감사 추적이 쉬워진다. 성능 측면에서는 벡터화, dtype 축소, 카테고리형 적용처럼 단일 노드 최적화를 먼저 적용하고, 이후 청크 처리와 병렬화를 검토한다.

pivot과 pivot_table을 고를 때도 조건이 다르다. pivot은 중복 키를 허용하지 않는 엄격성이 있고, pivot_table은 집계를 허용하는 대신 메모리 사용량까지 함께 고려해야 한다.

처리와 학습이 이어지는 예시

⤢✕

전제조건은 다음과 같다.

• Python 3.10+ / pandas 2.1+ / numpy 1.26+ / scikit-learn 1.3+ (환경별 최신 정보 확인 필요)
• 단일 파일 실행 가능, 인터넷 불필요

# -*- coding: utf-8 -*-
# sample_pipeline.py
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.compose import ColumnTransformer
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report

# 1) 데이터 생성/로드 + 스키마 검증
rng = pd.date_range("2024-01-01", periods=30, freq="D")
np.random.seed(42)

df = pd.DataFrame({
    "date": np.random.choice(rng, 500),
    "region": np.random.choice(["A", "B", "C"], 500),
    "product": np.random.choice(["P1", "P2"], 500),
    "qty": np.random.randint(1, 10, 500),
    "price": np.random.uniform(10, 100, 500).round(2)
})

# 타입 캐스팅 및 기본 검증
required = {"date": "datetime64[ns]", "region": "object", "product": "object",
            "qty": "int64", "price": "float64"}
for col, typ in required.items():
    if col not in df.columns:
        raise ValueError(f"누락 컬럼: {col}")
df = df.astype({"date": "datetime64[ns]", "region": "category",
                "product": "category"})  # 범주형 메모리 최적화

# 2) NumPy 벡터화 파생 변수
# 매출, 할인율, 원가 가정 및 마진 계산
revenue = df["qty"].to_numpy() * df["price"].to_numpy()
discount_rate = np.where(df["qty"].to_numpy() >= 8, 0.10, 0.02)
cost = df["price"].to_numpy() * 0.6  # 원가율 60% 가정
margin = revenue - (cost * df["qty"].to_numpy())
df["revenue"] = revenue * (1 - discount_rate)
df["margin"] = margin

# 3) GroupBy 집계
daily_region = (
    df.groupby(["date", "region"])
      .agg(
          qty_sum=("qty", "sum"),
          revenue_sum=("revenue", "sum"),
          price_mean=("price", "mean")
      )
      .reset_index()
)

# 4) pivot_table 리포트
pivot_rev = pd.pivot_table(
    daily_region,
    index="date",
    columns="region",
    values="revenue_sum",
    aggfunc="sum",
    fill_value=0,
    margins=False
).sort_index()

# 5) 분류 목표 생성: 일별 총매출 상위 30%를 High로 레이블링
total_rev = pivot_rev.sum(axis=1)
threshold = np.quantile(total_rev, 0.7)
y = (total_rev >= threshold).astype(int)  # 1: High, 0: Other
X = pivot_rev  # 특성: 지역별 일매출 피벗 컬럼

# 6) 학습/검증 파이프라인
num_features = list(X.columns)  # 전부 수치형
preprocess = ColumnTransformer(
    transformers=[
        ("num", StandardScaler(with_mean=True), num_features),
    ],
    remainder="drop"
)

clf = Pipeline(steps=[
    ("prep", preprocess),
    ("model", LogisticRegression(max_iter=500, n_jobs=None, random_state=42))
])

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, shuffle=True, random_state=42, stratify=y
)

# 교차검증 + 학습
cv_scores = cross_val_score(clf, X_train, y_train, cv=5, scoring="f1")
clf.fit(X_train, y_train)
pred = clf.predict(X_test)

print(f"CV F1 평균: {cv_scores.mean():.3f} ± {cv_scores.std():.3f}")
print(classification_report(y_test, pred))

set_output(transform="pandas")를 사용하면 전처리 뒤의 특성명을 유지할 수 있다. Pipeline과 cross_val_score를 함께 쓰면 데이터 누수를 방지할 수 있으며, pivot_table은 집계 과정의 결측을 fill_value=0으로 안전하게 채우는 방식으로 사용할 수 있다.

도구별 역할과 선택 기준

기법/도구	성능	확장성	일관성	운영 편의
파이썬 for-루프	하	하	중	상(직관)
NumPy 벡터화	상	중~상(메모리 의존)	상	중(브로드캐스팅 이해 필요)
Pandas GroupBy/Pivot	중~상	중(단일 노드 한계)	상(집계 규칙 고정)	상(표형 데이터 친화)
Scikit-learn Pipeline	중	중(병렬 CV로 확장)	상(누수 방지)	상(재현·배포 용이)

벡터화에서는 성능 향상과 메모리 사용량 증가 사이의 균형이 필요하다. 피벗 작업에서는 중복 키를 허용하지 않는 pivot의 엄격성과 집계를 허용하는 pivot_table의 유연성 중 무엇이 필요한지 판단해야 한다. Pipeline 캐시(joblib)를 활성화하면 디스크 I/O 비용이 늘어날 수 있다.

전처리·학습 자동화는 재실행 시간을 30~60% 단축할 수 있고, 교차검증과 파이프라인을 일원화하면 재현 실패율을 낮출 수 있다. 배포 실패율 50% 이상 저감 사례도 다수 있다. 스키마와 로깅을 표준화하면 디버깅이 쉬워지고, 특성 생성·집계 로직의 문서화와 감사 추적성을 확보할 수 있다. 데이터 누수를 막는 구조는 모델 신뢰도에도 직접 연결된다.