NYC 택시 수요 예측 파이프라인 만들기 (8) Airflow 자동화, 모니터링, CI

남은 퍼즐 세 개

7편까지로 수집 → 집계 → 학습 → 서빙이 전부 동작합니다. 다만 아직 매 단계를 사람이 명령어로 실행합니다. 이번 글에서 남은 세 조각을 채웁니다.

1. 자동화: 매달 반복될 수집, 집계, 적재를 Airflow DAG로
2. 모니터링: 데이터가 변하는 것과 서비스가 아픈 것을 알아채기
3. 품질: 설계 결정을 테스트로 고정하고 CI로 지키기

Airflow: 수동 명령을 그대로 DAG로

DAG 설계 원칙은 하나입니다. 태스크의 명령은 지금까지 터미널에서 치던 CLI와 동일하게 유지하고, Airflow는 순서, 재시도, 기록만 담당합니다.

flowchart LR
    A[download_raw] --> B[aggregate_hourly]
    B --> C[load_features_to_postgres]
    A --> D[upload_raw_to_minio]

aggregate_hourly = BashOperator(
    task_id="aggregate_hourly",
    bash_command=CD + f"python -m src.data.aggregate --month {MONTH}",
)

이렇게 하면 파이프라인 로직이 Airflow에 묶이지 않습니다. DAG가 깨져도 같은 명령을 손으로 실행할 수 있고, 각 모듈은 Airflow 없이 단독으로 테스트됩니다.

설계 포인트 몇 가지:

• 모든 태스크가 멱등입니다. MinIO 업로드는 같은 키 덮어쓰기, PostgreSQL 적재는 월 단위 삭제 후 삽입(6편에서 미리 깔아둔 설계)입니다. 실패한 태스크를 몇 번 재시도해도 데이터가 중복되거나 어긋나지 않습니다. 자동화의 안전은 스케줄러 설정이 아니라 태스크의 멱등성에서 나옵니다.
• max_active_runs=1: 같은 테이블에 두 run이 동시에 적재하면 기본 키 충돌이 나므로 동시 실행을 막았습니다.
• @monthly 스케줄 + 수동 실행 시 월 지정: TLC 데이터가 월 단위로 나오므로 스케줄도 월 단위입니다. 과거 달을 다시 처리하고 싶으면 {"month": "2023-12"}처럼 파라미터로 지정합니다.
• Airflow는 컨테이너로만 구동합니다. 의존성 충돌이 심한 패키지라 프로젝트 가상환경에 설치하지 않는다는 2편의 결정이 여기서 실행됐습니다. 컨테이너 시작 시 태스크용 패키지를 설치하는 _PIP_ADDITIONAL_REQUIREMENTS는 개발 편의 기능이라 프로덕션에서는 이미지를 직접 빌드해야 합니다.

모니터링: 두 개의 축

모니터링할 대상은 성격이 다른 두 가지입니다.

첫째, 데이터 드리프트. 모델은 “과거의 패턴이 반복된다”는 전제 위에 서 있습니다. 입력 분포가 변하면(계절 변화, 도시 행사, 요금 정책 변경) 그 전제가 무너지므로, 분포 변화 자체를 재학습 신호로 삼아야 합니다. Evidently로 기준 구간(학습 데이터)과 최근 구간의 피처와 타깃 분포를 비교하는 HTML 리포트를 만들었습니다. 지금은 수동 실행이지만, 구조상 Airflow 태스크로 넣어 매달 자동 생성하는 것이 다음 단계입니다.

둘째, 서비스 메트릭. 모델이 정확해도 API가 느리거나 죽어 있으면 소용이 없습니다. 7편에서 열어둔 /metrics 엔드포인트를 Prometheus가 주기적으로 긁어가고 Grafana로 봅니다. 요청량, 지연시간, 에러율이 기본입니다.

아직 남은 것도 명확합니다. Grafana 대시보드는 데이터소스만 연결된 상태라 패널 구성이 남았고, 모델 성능 추적(예측 vs 실제)은 실측값이 다음 달 데이터가 와야 생기는 지연 라벨 문제라 재학습 루프와 함께 설계할 예정입니다.

테스트: 설계 결정을 코드로 고정

pytest 19개를 만들었는데, 단순히 “함수가 안 죽는지”가 아니라 시리즈 내내 내린 설계 결정이 유지되는지를 검증합니다. 가장 아끼는 테스트는 데이터 누수 테스트입니다.

# pickup_count = 날짜값인 가짜 데이터를 만든다
# → 피처가 현재나 미래를 조금이라도 반영하면 값으로 바로 들통난다
row = at(out, "2024-01-09 06:00")
assert row["lag_24h"] == 8     # 하루 전 (1/8)
assert row["week_mean"] == 5   # 1/2~1/8 평균, 당일(9) 미포함

수요값을 날짜와 같게 만들어두면 “week_mean이 당일을 포함하는지”가 산술로 판별됩니다. 누수는 성능 지표로는 눈치채기 어렵지만(오히려 점수가 좋아집니다) 이런 테스트에는 즉시 걸립니다. 이 밖에 분할 경계, 그리드 계약, 서빙 피처와 학습 스키마의 일치가 전부 테스트로 고정되어 있습니다.

CI: 지키는 것은 기계가

GitHub Actions 워크플로는 단순합니다. push와 PR마다 uv sync --frozen으로 uv.lock 그대로 환경을 재현하고, ruff 린트와 pytest를 돌립니다. --frozen이 핵심인데, lock 파일과 다른 버전이 설치되는 것을 막아 “내 로컬에서는 됐는데” 문제를 원천 차단합니다.

시리즈를 마치며

1편에서 그린 아키텍처 다이어그램의 상자들이 이제 전부 실제 코드와 컨테이너로 존재합니다. 모델 학습에서 끝나던 프로젝트와 무엇이 달랐는지 돌아보면:

• 코드의 절반 이상이 모델이 아니라 데이터의 이동과 계약(집계, 적재, 서빙 피처 재현)에 쓰였습니다.
• 성능을 올린 것은 복잡한 모델이 아니라 공정한 비교 체계(Baseline, 분할 규칙, 봉인된 테스트)였습니다.
• 자동화를 쉽게 만든 것은 Airflow가 아니라 각 단계의 멱등성이었습니다.

남은 확장은 여러 달 데이터로의 스케일업, 모니터링과 재학습을 잇는 루프, 그리고 AWS 배포(MinIO → S3, RDS, ECS)입니다. 진행되는 대로 이 시리즈에 이어 쓰겠습니다.