Kubeflow 없이 GitHub Actions로 최소 MLOps 구축

.
├── src/
│   ├── data.py
│   ├── train.py
│   ├── evaluate.py
│   └── infer.py
├── pipelines/
│   ├── validate_data.py
│   ├── run_train.py
│   ├── run_eval.py
│   └── package_model.py
├── configs/
│   └── train.yaml
├── model_card/
│   └── README.md
└── pyproject.toml

# pipelines/validate_data.py
import json
import pandas as pd

REQUIRED_COLS = {
    "user_id": "int64",
    "feature_a": "float64",
    "feature_b": "float64",
    "label": "int64",
}


def main(path: str) -> None:
    df = pd.read_parquet(path)

    missing = [c for c in REQUIRED_COLS if c not in df.columns]
    if missing:
        raise SystemExit(f"Missing columns: {missing}")

    for col, dtype in REQUIRED_COLS.items():
        if str(df[col].dtype) != dtype:
            raise SystemExit(f"Type mismatch: {col} is {df[col].dtype}, expected {dtype}")

    na_rate = df.isna().mean().to_dict()
    if na_rate["label"] > 0:
        raise SystemExit("Label contains NA")

    if na_rate["feature_a"] > 0.2:
        raise SystemExit(f"Too many NA in feature_a: {na_rate['feature_a']}")

    report = {
        "rows": int(len(df)),
        "na_rate": na_rate,
        "dtypes": {c: str(df[c].dtype) for c in REQUIRED_COLS},
    }
    print(json.dumps(report, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser()
    p.add_argument("--data", required=True)
    args = p.parse_args()

    main(args.data)

# pipelines/run_eval.py
import json

THRESHOLDS = {
    "auc": 0.80,
    "f1": 0.60,
}


def main(metrics_path: str) -> None:
    with open(metrics_path, "r", encoding="utf-8") as f:
        metrics = json.load(f)

    for k, v in THRESHOLDS.items():
        if metrics.get(k, 0) < v:
            raise SystemExit(f"Metric gate failed: {k}={metrics.get(k)} < {v}")

    print("Metric gate passed")


if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser()
    p.add_argument("--metrics", required=True)
    args = p.parse_args()

    main(args.metrics)

# .github/workflows/mlops.yml
name: minimal-mlops

on:
  pull_request:
  push:
    branches: ["main"]
  schedule:
    - cron: "0 2 * * *"
  workflow_dispatch:

jobs:
  test-and-validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install
        run: |
          python -m pip install -U pip
          pip install -r requirements.txt

      - name: Unit tests
        run: pytest -q

      - name: Validate data
        run: |
          python pipelines/validate_data.py --data data/train.parquet

  train-eval-package:
    if: github.ref == 'refs/heads/main'
    needs: [test-and-validate]
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Restore pip cache
        uses: actions/cache@v4
        with:
          path: |
            ~/.cache/pip
          key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }}
          restore-keys: |
            ${{ runner.os }}-pip-

      - name: Install
        run: |
          python -m pip install -U pip
          pip install -r requirements.txt

      - name: Train
        run: |
          python pipelines/run_train.py --config configs/train.yaml --out artifacts

      - name: Evaluate (metric gate)
        run: |
          python pipelines/run_eval.py --metrics artifacts/metrics.json

      - name: Package
        run: |
          python pipelines/package_model.py --in artifacts --out dist

      - name: Upload artifact
        uses: actions/upload-artifact@v4
        with:
          name: model-dist
          path: dist

# pipelines/package_model.py
import hashlib
import json
from pathlib import Path


def sha256_file(p: Path) -> str:
    h = hashlib.sha256()
    with p.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def main(inp: str, out: str) -> None:
    inp_dir = Path(inp)
    out_dir = Path(out)
    out_dir.mkdir(parents=True, exist_ok=True)

    model_path = inp_dir / "model.pkl"
    metrics_path = inp_dir / "metrics.json"

    meta = {
        "model_sha256": sha256_file(model_path),
        "metrics": json.loads(metrics_path.read_text(encoding="utf-8")),
    }

    (out_dir / "metadata.json").write_text(
        json.dumps(meta, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

    # 실제 환경에서는 model.pkl, tokenizer.json 같은 파일도 함께 복사
    (out_dir / "model.pkl").write_bytes(model_path.read_bytes())


if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser()
    p.add_argument("--in", dest="inp", required=True)
    p.add_argument("--out", dest="out", required=True)
    args = p.parse_args()

    main(args.inp, args.out)

deploy-prod:
  needs: [train-eval-package]
  runs-on: ubuntu-latest
  environment: production
  steps:
    - uses: actions/download-artifact@v4
      with:
        name: model-dist
        path: dist

    - name: Deploy
      run: |
        ./scripts/deploy.sh dist