MLOps CI/CD for Machine Learning

Real-World Scenario

A fintech ML team deploys a credit scoring model. Before CI/CD, every deployment was manual: a data scientist ran a notebook, emailed results to an engineer, who then deployed via SSH. One night a corrupted training batch went unnoticed, and a degraded model ran in production for 6 hours. After implementing ML CI/CD, data quality checks catch schema errors, model gates block any model below the performance threshold, and the entire pipeline runs on every push.

ML Pipeline Structure

Push to main
    ↓
1. Data validation   — schema check, null rates, distribution tests
    ↓
2. Model training    — reproducible training script
    ↓
3. Model evaluation  — metrics on holdout set
    ↓
4. Model gating      — compare vs baseline; fail if below threshold
    ↓
5. Model registry    — register version in MLflow
    ↓
6. Deployment        — serve new version (blue/green or canary)

GitHub Actions Workflow

# .github/workflows/ml-pipeline.yml
name: ML Training Pipeline

on:
  push:
    branches: [main]
    paths:
      - 'src/**'
      - 'data/**'
      - 'requirements.txt'

jobs:
  validate-data:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements.txt
      - name: Validate data quality
        run: python scripts/validate_data.py
        env:
          DATA_PATH: data/training.csv

  train-and-evaluate:
    needs: validate-data
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      - run: pip install -r requirements.txt
      - name: Train model
        run: python scripts/train.py --output-dir models/
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }}
      - name: Evaluate and gate
        run: python scripts/evaluate_and_gate.py
        env:
          MIN_ACCURACY: 0.90
          MIN_AUC:      0.88
      - name: Upload model artifact
        uses: actions/upload-artifact@v4
        with:
          name: trained-model
          path: models/

  deploy:
    needs: train-and-evaluate
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v4
      - name: Download model
        uses: actions/download-artifact@v4
        with:
          name: trained-model
          path: models/
      - name: Deploy to staging
        run: python scripts/deploy.py --env staging
        env:
          DEPLOY_KEY: ${{ secrets.DEPLOY_KEY }}

Data Validation Script

# scripts/validate_data.py
import sys
import json
import argparse
import numpy as np
import pandas as pd
from pathlib import Path

SCHEMA = {
    "required_columns": ["age", "income", "credit_score", "debt_ratio", "label"],
    "dtypes": {
        "age":          "numeric",
        "income":       "numeric",
        "credit_score": "numeric",
        "debt_ratio":   "numeric",
        "label":        "numeric",
    },
    "ranges": {
        "age":          (18, 120),
        "credit_score": (300, 850),
        "debt_ratio":   (0.0, 10.0),
    },
    "max_null_rate":     0.05,   # fail if any column has > 5% nulls
    "min_rows":          1000,
}

def validate(data_path: str) -> dict:
    issues = []
    df = pd.read_csv(data_path)

    # Row count
    if len(df) < SCHEMA["min_rows"]:
        issues.append(f"Too few rows: {len(df)} < {SCHEMA['min_rows']}")

    # Required columns
    missing_cols = set(SCHEMA["required_columns"]) - set(df.columns)
    if missing_cols:
        issues.append(f"Missing columns: {missing_cols}")
        return {"passed": False, "issues": issues}  # can't continue without columns

    # Null rates
    for col in SCHEMA["required_columns"]:
        null_rate = df[col].isnull().mean()
        if null_rate > SCHEMA["max_null_rate"]:
            issues.append(f"Column '{col}' null rate {null_rate:.1%} > {SCHEMA['max_null_rate']:.1%}")

    # Numeric type check
    for col, expected in SCHEMA["dtypes"].items():
        if col in df.columns and expected == "numeric":
            if not pd.api.types.is_numeric_dtype(df[col]):
                issues.append(f"Column '{col}' is not numeric: {df[col].dtype}")

    # Range checks
    for col, (lo, hi) in SCHEMA["ranges"].items():
        if col in df.columns:
            out_of_range = ((df[col] < lo) | (df[col] > hi)).sum()
            if out_of_range > 0:
                issues.append(f"Column '{col}': {out_of_range} values outside [{lo}, {hi}]")

    # Class balance check for label
    if "label" in df.columns:
        class_counts = df["label"].value_counts(normalize=True)
        if class_counts.min() < 0.02:
            issues.append(f"Severe class imbalance: minority class = {class_counts.min():.1%}")

    return {"passed": len(issues) == 0, "issues": issues, "n_rows": len(df)}


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--data-path", default="data/training.csv")
    args = parser.parse_args()

    result = validate(args.data_path)
    print(json.dumps(result, indent=2))

    if not result["passed"]:
        print("\n❌ DATA VALIDATION FAILED", file=sys.stderr)
        for issue in result["issues"]:
            print(f"  - {issue}", file=sys.stderr)
        sys.exit(1)

    print(f"\n✅ Data validation passed ({result['n_rows']:,} rows)")

Model Gating Script

# scripts/evaluate_and_gate.py
import sys
import json
import os
import joblib
import numpy as np
import mlflow
from sklearn.metrics import accuracy_score, roc_auc_score, f1_score
from sklearn.model_selection import train_test_split

# Thresholds from environment variables (set in CI/CD)
MIN_ACCURACY = float(os.environ.get("MIN_ACCURACY", 0.90))
MIN_AUC      = float(os.environ.get("MIN_AUC",      0.88))
MIN_F1       = float(os.environ.get("MIN_F1",       0.85))


def evaluate_model(model_path: str, data_path: str) -> dict:
    import pandas as pd
    df = pd.read_csv(data_path)
    X = df.drop("label", axis=1).values
    y = df["label"].values

    _, X_test, _, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)

    model = joblib.load(model_path)
    y_pred = model.predict(X_test)
    y_prob = model.predict_proba(X_test)[:, 1]

    return {
        "accuracy": accuracy_score(y_test, y_pred),
        "auc_roc":  roc_auc_score(y_test, y_prob),
        "f1":       f1_score(y_test, y_pred),
        "n_test":   len(y_test),
    }


def gate_model(metrics: dict) -> tuple[bool, list[str]]:
    """Return (passed, list_of_failures)."""
    failures = []
    if metrics["accuracy"] < MIN_ACCURACY:
        failures.append(f"Accuracy {metrics['accuracy']:.4f} < threshold {MIN_ACCURACY}")
    if metrics["auc_roc"] < MIN_AUC:
        failures.append(f"AUC-ROC {metrics['auc_roc']:.4f} < threshold {MIN_AUC}")
    if metrics["f1"] < MIN_F1:
        failures.append(f"F1 {metrics['f1']:.4f} < threshold {MIN_F1}")
    return len(failures) == 0, failures


if __name__ == "__main__":
    metrics = evaluate_model("models/model.pkl", "data/training.csv")
    passed, failures = gate_model(metrics)

    print("Model Evaluation Results:")
    for k, v in metrics.items():
        print(f"  {k:12s}: {v:.4f}" if isinstance(v, float) else f"  {k:12s}: {v}")

    print(f"\nThresholds: accuracy≥{MIN_ACCURACY}, AUC≥{MIN_AUC}, F1≥{MIN_F1}")

    if passed:
        print("\n✅ Model passed all gates — eligible for promotion")
        # Log to MLflow, register model
        with mlflow.start_run(run_name="CI-evaluation"):
            for name, val in metrics.items():
                if isinstance(val, float):
                    mlflow.log_metric(name, val)
    else:
        print("\n❌ MODEL GATE FAILED:", file=sys.stderr)
        for f in failures:
            print(f"  - {f}", file=sys.stderr)
        sys.exit(1)

DVC — Data Version Control

# Install DVC
pip install dvc dvc-s3

# Initialize in your ML repo
dvc init
git add .dvc .dvcignore
git commit -m "Initialize DVC"

# Track data file — creates data/training.csv.dvc (small file, goes to git)
dvc add data/training.csv
git add data/training.csv.dvc data/.gitignore
git commit -m "Track training data with DVC"

# Push data to S3 (configure remote first)
dvc remote add -d myremote s3://my-bucket/dvc-store
dvc push

# On another machine: pull the exact data version
git clone <repo>
dvc pull  # downloads data from S3

# Create a pipeline stage (cached, only reruns when inputs change)
dvc run -n train \
    -d data/training.csv -d src/train.py \
    -o models/model.pkl \
    -m metrics.json \
    python src/train.py