실험 데이터 클리닝 파이프라인 — 결측·중복·이상치를 재현 가능하게 처리

이 토픽을 마치면

교과서에서 배운 데이터 클리닝, apply/map, mutable 인자 함정 을 엮어서, 지저분한 실험 데이터 CSV를 재현 가능한 방식으로 정제하는 파이프라인을 직접 만들 수 있습니다. 매번 다르게 클리닝하다가 결과가 재현되지 않던 문제가 사라집니다.

이 글은 교육용 일반 예제 입니다. 실전은 dbt, Great Expectations, Pandera 같은 정교한 도구를 씁니다.

"지난주 결과가 왜 안 나오지?" — 수동 클리닝의 함정

여러분이 매주 실험 데이터 CSV를 받고 분석한다고 합시다. 첫 주:

Excel로 열어 결측 셀 눈으로 확인
이상해 보이는 값 삭제
여러 파일에서 나온 것 하나로 합치기
저장하고 분석

다음 주 같은 과정을 반복합니다. 그리고 여러분의 지도교수가 물어봅니다 "지난주 결과가 왜 이번 주에 재현이 안 되지?"

원인이 하나만 있는 게 아닙니다.

문제 1: 눈으로 지운 값이 매번 다릅니다. 어떤 주는 3σ 밖만 지우고 다른 주는 4σ 밖만 지웁니다.

문제 2: 결측 처리 방식이 매번 다릅니다. 어떤 주는 지우고 어떤 주는 평균으로 채웁니다.

문제 3: 어떤 파일 몇 개가 이번에 빠졌는지 기억이 안 납니다.

진짜 접근은 명시적 파이프라인입니다. 각 클리닝 스텝을 코드로 명세하고, 그 코드를 저장하고 버전 관리합니다. 다시 같은 코드로 같은 데이터를 처리하면 정확히 같은 결과가 나옵니다. 이것이 과학적 재현성의 기본입니다.

블랙박스에서 부품으로

부품 1: pandas의 명시적 스텝

각 클리닝 액션을 별개 함수로.

python

import pandas as pd
import numpy as np


def load_and_validate(path: str, required_cols: list[str]) -> pd.DataFrame:
    df = pd.read_csv(path)
    missing = set(required_cols) - set(df.columns)
    if missing:
        raise ValueError(f"Missing columns: {missing}")
    return df


def remove_duplicates(df: pd.DataFrame, subset: list[str]) -> tuple[pd.DataFrame, int]:
    n_before = len(df)
    df_clean = df.drop_duplicates(subset=subset).copy()
    return df_clean, n_before - len(df_clean)


def drop_missing(df: pd.DataFrame, required_cols: list[str]) -> tuple[pd.DataFrame, int]:
    n_before = len(df)
    df_clean = df.dropna(subset=required_cols).copy()
    return df_clean, n_before - len(df_clean)


def filter_outliers_iqr(df: pd.DataFrame, col: str, factor: float = 1.5) -> tuple[pd.DataFrame, int]:
    q1 = df[col].quantile(0.25)
    q3 = df[col].quantile(0.75)
    iqr = q3 - q1
    low, high = q1 - factor * iqr, q3 + factor * iqr
    n_before = len(df)
    df_clean = df[(df[col] >= low) & (df[col] <= high)].copy()
    return df_clean, n_before - len(df_clean)

핵심: 각 함수가 얼마나 제거했는지(count) 를 반환. 이것을 나중에 로그로 남깁니다.

부품 2: apply와 map으로 컬럼 변환

값 변환은 apply 나 map 으로 명시적으로.

python

def normalize_gene_symbols(df: pd.DataFrame, col: str = "gene") -> pd.DataFrame:
    """
    유전자 심볼을 대문자 표준으로 변환.
    """
    df = df.copy()
    df[col] = df[col].str.upper().str.strip()
    return df


def standardize_units(df: pd.DataFrame, col: str, unit_col: str) -> pd.DataFrame:
    """
    농도 값과 단위를 조합해 μM로 통일.
    """
    df = df.copy()
    
    def to_micromolar(row):
        value = row[col]
        unit = row[unit_col].lower()
        multiplier = {"nm": 0.001, "μm": 1, "um": 1, "mm": 1000, "m": 1_000_000}
        return value * multiplier.get(unit, np.nan)
    
    df["concentration_uM"] = df.apply(to_micromolar, axis=1)
    return df

주의: 항상 df.copy(). 원본 DataFrame 변경 방지.

부품 3: mutable 함정 회피

파이썬 초보자의 대표 실수: 가변 기본값.

python

# 위험한 코드
def add_error_log(errors: list = []) -> list:
    errors.append("something wrong")
    return errors


# 첫 호출: ["something wrong"]
# 두 번째 호출: ["something wrong", "something wrong"] — 예상 밖!

이유: 파이썬 함수의 기본값은 함수 정의 시 한 번만 평가됩니다. errors=[] 의 그 [] 는 함수가 정의될 때 만들어져 계속 재사용됩니다.

정정:

python

def add_error_log(errors: list | None = None) -> list:
    if errors is None:
        errors = []
    errors.append("something wrong")
    return errors

같은 함정이 DataFrame에도 있습니다.

python

# 위험
def process(df):
    df["new_col"] = df["old_col"] * 2  # 원본 df 변경!
    return df

original = pd.read_csv("data.csv")
processed = process(original)
# original["new_col"] 도 존재. 원본이 오염됨.


# 안전
def process(df):
    df = df.copy()
    df["new_col"] = df["old_col"] * 2
    return df

파이프라인의 각 스텝이 순수 함수(입력을 변경하지 않고 새 출력 반환)여야 재현성이 유지됩니다.

파이프라인 조립

이제 위 부품들을 하나의 파이프라인으로.

python

from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class CleaningReport:
    input_rows: int = 0
    output_rows: int = 0
    steps: list = field(default_factory=list)
    
    def add(self, step_name: str, removed: int, details: str = "") -> None:
        self.steps.append({
            "step": step_name,
            "removed": removed,
            "details": details,
            "timestamp": datetime.now().isoformat()
        })


def clean_experiment_data(
    csv_path: str,
    required_cols: list[str] = None,
    outlier_col: str = "value",
    outlier_factor: float = 1.5
) -> tuple[pd.DataFrame, CleaningReport]:
    required_cols = required_cols or ["gene", "value", "unit"]
    
    df = load_and_validate(csv_path, required_cols)
    report = CleaningReport(input_rows=len(df))
    
    df = normalize_gene_symbols(df, "gene")
    report.add("normalize_gene_symbols", 0, "uppercased and stripped")
    
    df, n_dup = remove_duplicates(df, subset=["gene", "value"])
    report.add("remove_duplicates", n_dup)
    
    df, n_missing = drop_missing(df, required_cols)
    report.add("drop_missing", n_missing)
    
    df = standardize_units(df, "value", "unit")
    report.add("standardize_units", 0)
    
    df, n_outlier = filter_outliers_iqr(df, outlier_col, outlier_factor)
    report.add("filter_outliers_iqr", n_outlier, f"factor={outlier_factor}")
    
    report.output_rows = len(df)
    return df, report

사용:

python

clean_df, report = clean_experiment_data("raw_data.csv")

print(f"입력: {report.input_rows}행, 출력: {report.output_rows}행")
for step in report.steps:
    print(f"  {step['step']}: -{step['removed']} ({step['details']})")

기대 출력:

text

입력: 1000행, 출력: 934행
  normalize_gene_symbols: -0 (uppercased and stripped)
  remove_duplicates: -12
  drop_missing: -30
  standardize_units: -0
  filter_outliers_iqr: -24 (factor=1.5)

페이딩 — 여러분이 채워야 할 세 빈칸

빈칸 1: 스키마 검증

각 컬럼의 타입과 범위를 명시적으로 검증합니다.

python

def validate_schema(df: pd.DataFrame, schema: dict) -> list[str]:
    """
    schema = {"col_name": {"type": float, "min": 0, "max": 100}}
    검증 실패 항목의 에러 메시지 리스트 반환. 빈 리스트면 모두 통과.
    """
    errors = []
    
    for col, rules in schema.items():
        if col not in df.columns:
            errors.append(f"Missing column: {col}")
            continue
        
        # TODO 1: 타입 검증 (df[col].dtype 확인)
        # TODO 2: min/max 범위 검증 (있으면 df[col].min(), df[col].max() 비교)
        pass
    
    return errors

힌트: if not pd.api.types.is_numeric_dtype(df[col]): errors.append(...). 범위: if "min" in rules and df[col].min() < rules["min"]: errors.append(...).

빈칸 2: 로그 파일 저장

CleaningReport를 CSV 로그로 저장.

python

def save_report(report: CleaningReport, log_path: str) -> None:
    """
    report.steps 를 CSV로 저장.
    """
    # TODO: pd.DataFrame(report.steps) 로 변환 후 to_csv
    pass

빈칸 3: 재현성 해시

동일 입력에 동일 코드가 동일 출력을 만드는지 확인하는 해시.

python

import hashlib


def compute_output_hash(df: pd.DataFrame) -> str:
    """
    df의 내용을 정렬 후 해시.
    같은 df에서 항상 같은 해시가 나와야 함.
    """
    # TODO 1: df를 안정적 순서(예: 모든 컬럼으로 정렬)로 정렬
    # TODO 2: to_csv(index=False) 로 문자열화
    # TODO 3: hashlib.sha256(bytes).hexdigest() 반환
    pass

힌트:

python

sorted_df = df.sort_values(list(df.columns)).reset_index(drop=True)
content = sorted_df.to_csv(index=False).encode()
return hashlib.sha256(content).hexdigest()[:16]

성찰 — 실전 데이터 파이프라인과의 차이

dbt (data build tool): SQL 기반 파이프라인의 표준. 각 스텝이 SQL model, 의존성이 자동 관리, 테스트가 파이프라인에 통합. Airbnb/Netflix 같은 곳의 표준.

Great Expectations / Pandera: 데이터 스키마·품질 검증 프레임워크. 각 컬럼의 기대치를 선언적으로 명세하고 위반 시 알림.

Apache Airflow / Prefect: 대규모 파이프라인 오케스트레이션. 스케줄링, 재시도, 백필 등의 기능.

dask / polars: 여러분의 데이터가 100GB 이상이면 pandas 대신 이 도구들. 병렬 처리와 lazy evaluation.

data lineage 추적: 실전은 각 출력이 어떤 입력에서 어떤 변환을 거쳤는지 자동 추적. OpenLineage 같은 표준.

확장 프로젝트

1. Streamlit 대시보드: 여러분의 클리닝 파이프라인을 웹 UI로. 사용자 CSV 업로드 → 리포트 시각화.

2. Great Expectations 통합: 여러분의 CleaningReport 대신 Great Expectations의 표준 리포트로.

3. 여러 파일 병합: 여러 CSV를 합칠 때 스키마 정합성 자동 검증 후 병합.

4. 데이터 카탈로그: 각 실험 데이터의 메타데이터를 SQLite에 저장하고 검색.

이 편의 부품 지도

[F] 데이터 클리닝: 결측·중복·이상치의 명시적 처리. 각 스텝의 로그.
[F] apply/map: 컬럼 변환의 함수형 접근. axis=1 이해.
[F] mutable 함정: 가변 기본값, DataFrame in-place 변경 회피. 순수 함수 원칙.
[W] 파일 I/O: CSV 읽기·쓰기 (완성 스크립트 제공).

[F] = 여러분이 직접 구현 / [W] = 완성 코드로 제공.