NYC 택시 수요 예측 파이프라인 만들기 (6) PostgreSQL, MinIO, MLflow 연동

로컬 파일로 돌아가던 파이프라인에 오브젝트 스토리지, 피처 DB, 실험 추적 서버를 연결한 과정을 정리했습니다.

Posted 2026/07/17

By 강호승

3 min read

왜 파일만으로는 부족한가

5편까지의 작업은 사실 로컬 parquet 파일만으로 전부 돌아갑니다. 그런데 이 프로젝트의 목표는 학습에서 끝나지 않는 파이프라인이고, 다음 단계들이 파일만으로는 안 됩니다.

서빙: 예측 API가 요청마다 과거 수요를 조회해야 하는데, 48MB parquet을 매 요청마다 읽을 수는 없습니다 → PostgreSQL
원본 보관: Airflow가 매달 새 데이터를 수집하면 어딘가에 쌓아야 합니다. 로컬 디스크가 아니라 S3처럼 쓸 수 있는 곳에 → MinIO
실험 관리: 모델, 파라미터, 지표가 실험마다 쌓이는데, 터미널 출력 복사로는 일주일이면 뒤섞입니다 → MLflow

이번 글은 이 세 저장소를 Docker Compose로 띄우고 파이프라인 코드와 연결한 기록입니다.

Docker 스택 시동

2편에서 계획만 세워둔 스택 중 저장소 세 개를 먼저 띄웁니다.

서비스	역할	포인트
postgres:16	피처 테이블 + MLflow/Airflow 메타 DB	초기화 SQL로 DB 여러 개 생성
minio	S3 호환 오브젝트 스토리지	원본 데이터 + MLflow 아티팩트
minio-init	버킷 생성 후 종료되는 1회성 컨테이너	`mc mb` 실행하고 exit
mlflow	실험 추적 서버	백엔드는 PG, 아티팩트는 MinIO

몇 가지 배운 것들:

healthcheck와 기동 순서: MLflow는 PostgreSQL이 준비된 뒤에 떠야 합니다. depends_on에 condition: service_healthy를 걸면 “컨테이너가 시작됨”이 아니라 “실제로 접속 가능함”을 기다립니다.
1회성 초기화 컨테이너 패턴: MinIO는 버킷을 미리 만들어주지 않으므로, minio/mc 이미지로 버킷만 만들고 종료하는 컨테이너를 둡니다. 수동으로 웹 콘솔에 들어가 버킷을 만드는 단계가 사라집니다.
PostgreSQL 초기화 스크립트: 컨테이너 최초 기동 시 /docker-entrypoint-initdb.d/의 SQL이 실행됩니다. 여기서 MLflow용 메타 DB를 추가로 만들어 PostgreSQL 하나로 여러 서비스의 DB를 겸용합니다.

접속 정보는 전부 .env로

코드 어디에도 주소나 비밀번호를 박지 않고, 저장소 클라이언트를 한 모듈로 모았습니다.

  
def s3_client():
    return boto3.client(
        "s3",
        endpoint_url=os.getenv("MINIO_ENDPOINT"),
        aws_access_key_id=os.getenv("MINIO_ACCESS_KEY"),
        aws_secret_access_key=os.getenv("MINIO_SECRET_KEY"),
    )

MinIO는 S3 호환 API라 클라이언트가 AWS 공식 SDK인 boto3 그대로입니다. 나중에 AWS로 옮길 때 코드는 그대로 두고 .env의 endpoint와 키만 바꾸면 됩니다. PostgreSQL 접속도 같은 방식입니다. 로컬 Docker에서 RDS로 옮겨도 코드 수정이 없습니다.

원본은 MinIO로: 키 구조는 공식 배포 경로 그대로

원본 parquet을 버킷에 올릴 때 키 구조를 TLC 공식 배포 경로(trip-data/, misc/)와 동일하게 잡았습니다. “우리 버킷은 공식 소스의 미러”라는 규칙이라 새로 외울 구조가 없고, 같은 키에 다시 올리면 덮어쓰기라 재실행해도 안전합니다.

피처는 PostgreSQL로: 멱등 적재

집계 테이블을 DB에 넣는 부분에는 설계 결정이 두 개 있습니다.

  
CREATE TABLE IF NOT EXISTS hourly_demand (
    pickup_hour   timestamp NOT NULL,
    pulocation_id smallint  NOT NULL,
    pickup_count  integer   NOT NULL,
    PRIMARY KEY (pickup_hour, pulocation_id)
)

첫째, 기본 키가 (시각, 지역)입니다. 4편에서 assert로 지키던 “한 칸에 정답 하나” 그리드 계약을 DB 수준에서 강제합니다. 코드에 버그가 생겨 중복 행을 넣으려 하면 DB가 거부합니다.

둘째, 월 단위 멱등 적재입니다. 해당 월의 기존 행을 지우고 다시 넣는 것을 한 트랜잭션으로 묶었습니다.

  
with engine.begin() as conn:  # 전부 성공하거나 전부 취소
    conn.execute(text("DELETE FROM hourly_demand WHERE pickup_hour >= :start AND pickup_hour < :end"), ...)
    demand.to_sql("hourly_demand", conn, if_exists="append", ...)

같은 달을 몇 번 실행해도 결과가 같고, 다른 달에는 영향이 없습니다. 행 단위 upsert보다 단순하고, “이 월의 데이터는 이 실행 결과가 전부”라는 보장이 명확합니다. 이 멱등성은 나중에 Airflow가 실패한 태스크를 재시도할 때 진가를 발휘합니다. 재실행이 안전하도록 태스크를 설계해두면 자동화 단계에서 고민할 것이 사라집니다.

실험은 MLflow로

MLflow 서버는 기록을 두 곳에 나눠 저장합니다. 파라미터와 지표 같은 메타데이터는 PostgreSQL에, 모델 파일 같은 아티팩트는 MinIO 버킷에 들어갑니다. 방금 띄운 저장소 두 개를 MLflow가 그대로 재사용하는 구성입니다.

학습 스크립트가 run마다 남기는 것:

하이퍼파라미터와 피처 목록
Baseline을 포함한 비교 지표 전체. run 하나만 열어도 “그때 Baseline 대비 얼마나 좋았는지”가 보입니다
학습된 모델. LightGBM은 MLflow 표준 포맷에 더해 네이티브 텍스트 포맷(lgbm_booster.txt)을 별도 아티팩트로 함께 저장합니다

네이티브 포맷을 이중으로 남기는 이유는 다음 글의 주인공입니다. 서빙 컨테이너가 피클 역직렬화 없이 모델을 불러오기 위해서입니다.

저장소가 모두 연결됐으니, 다음 글에서는 MLflow의 모델을 불러와 FastAPI 예측 API로 서빙합니다. 학습 때와 서빙 때의 피처를 어떻게 일치시키는지가 핵심입니다.

Projects, ML Pipeline