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 stage add -n train \
    -d data/training.csv -d src/train.py \
    -o models/model.pkl \
    -m metrics.json \
    python src/train.py
dvc repro   # runs the stage; skips it next time if no dependency changed

Practice

Do these in one scratch project with scripts/, data/, models/ and tests/ directories. Save the lesson’s two scripts as scripts/validate_data.py and scripts/evaluate_and_gate.py, unchanged. You need pandas, scikit-learn, joblib, mlflow and pytest. These run the pipeline’s steps locally, the same commands the workflow’s run: lines call. Nothing here runs on GitHub Actions.

1. The validator needs a data/training.csv to check. Write a seeded generator that produces 2,000 rows in the schema’s columns, with a label that depends on credit_score and debt_ratio. Then run the validator on it.

# scripts/make_data.py: a seeded synthetic credit dataset in the validator's schema
import numpy as np
import pandas as pd

rng = np.random.default_rng(7)
n = 2000
df = pd.DataFrame({
    "age":          rng.integers(18, 80, n),
    "income":       rng.lognormal(10.8, 0.4, n).round(0),
    "credit_score": rng.integers(300, 851, n),
    "debt_ratio":   rng.uniform(0.0, 3.0, n).round(3),
})
risk = (df["credit_score"] - 575) / 60 - (df["debt_ratio"] - 1.5) * 2
df["label"] = (risk + rng.normal(0, 0.8, n) > 0).astype(int)
df.to_csv("data/training.csv", index=False)
print(f"wrote data/training.csv: {len(df)} rows, label mean {df['label'].mean():.3f}")

$ python scripts/make_data.py
wrote data/training.csv: 2000 rows, label mean 0.504
$ python scripts/validate_data.py
{
  "passed": true,
  "issues": [],
  "n_rows": 2000
}

✅ Data validation passed (2,000 rows)

2. Break the data on purpose and confirm the validator stops the pipeline. Copy the clean file to data/broken.csv, then make two changes: set credit_score to 900 in the first 25 rows, and blank income in every 15th row. Run the validator on the broken file and check the exit code.

# scripts/corrupt.py: two deliberate data bugs, written to a separate file
import numpy as np
import pandas as pd

df = pd.read_csv("data/training.csv")
df.loc[:24, "credit_score"] = 900          # 25 rows above the 850 maximum
df.loc[::15, "income"] = np.nan            # every 15th row loses its income
df.to_csv("data/broken.csv", index=False)
print("wrote data/broken.csv")

$ python scripts/corrupt.py
wrote data/broken.csv
$ python scripts/validate_data.py --data-path data/broken.csv
{
  "passed": false,
  "issues": [
    "Column 'income' null rate 6.7% > 5.0%",
    "Column 'credit_score': 25 values outside [300, 850]"
  ],
  "n_rows": 2000
}

❌ DATA VALIDATION FAILED
  - Column 'income' null rate 6.7% > 5.0%
  - Column 'credit_score': 25 values outside [300, 850]
$ echo $?
1

3. The gate script loads models/model.pkl, but the lesson never shows how it gets there. Write scripts/train.py so that it holds out exactly the rows the gate tests on. Then run the gate twice: once with default thresholds, and once with MIN_ACCURACY=0.99 set in the environment, the way the workflow’s env: block sets it.

# scripts/train.py: train on the same split evaluate_and_gate.py holds out
import joblib
import pandas as pd
from pathlib import Path
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

df = pd.read_csv("data/training.csv")
X, y = df.drop("label", axis=1).values, df["label"].values
X_train, _, y_train, _ = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)

model = make_pipeline(StandardScaler(), LogisticRegression())
model.fit(X_train, y_train)
Path("models").mkdir(exist_ok=True)
joblib.dump(model, "models/model.pkl")
print(f"trained on {len(X_train)} rows -> models/model.pkl")

$ python scripts/train.py
trained on 1600 rows -> models/model.pkl
$ python scripts/evaluate_and_gate.py
Model Evaluation Results:
  accuracy    : 0.9450
  auc_roc     : 0.9878
  f1          : 0.9455
  n_test      : 400

Thresholds: accuracy≥0.9, AUC≥0.88, F1≥0.85

✅ Model passed all gates — eligible for promotion
2026/10/11 11:16:19 INFO mlflow.store.db.utils: Creating initial MLflow database tables...
2026/10/11 11:16:19 INFO mlflow.store.db.utils: Updating database tables
$ MIN_ACCURACY=0.99 python scripts/evaluate_and_gate.py
Model Evaluation Results:
  accuracy    : 0.9450
  auc_roc     : 0.9878
  f1          : 0.9455
  n_test      : 400

Thresholds: accuracy≥0.99, AUC≥0.88, F1≥0.85

❌ MODEL GATE FAILED:
  - Accuracy 0.9450 < threshold 0.99
$ echo $?
1

4. CI should catch a broken gate before it ever sees a model. Write pytest tests that check three things: gate_model passes good metrics, it reports one failure per failing metric, and validate flags a single out-of-range credit_score.

# tests/test_pipeline.py: unit tests CI can run before any training happens
import sys
sys.path.insert(0, "scripts")

import pandas as pd
from validate_data import validate
from evaluate_and_gate import gate_model

GOOD = {"accuracy": 0.95, "auc_roc": 0.97, "f1": 0.94}

def test_gate_passes_good_model():
    assert gate_model(GOOD) == (True, [])

def test_gate_names_every_failing_metric():
    passed, failures = gate_model({"accuracy": 0.80, "auc_roc": 0.97, "f1": 0.70})
    assert not passed
    assert len(failures) == 2

def test_validate_flags_out_of_range_credit_score(tmp_path):
    df = pd.read_csv("data/training.csv")
    df.loc[0, "credit_score"] = 900
    path = tmp_path / "one_bad_row.csv"
    df.to_csv(path, index=False)
    result = validate(str(path))
    assert result["issues"] == ["Column 'credit_score': 1 values outside [300, 850]"]

$ python -m pytest -q tests/
...                                                                      [100%]
3 passed in 2.31s

5. The FAQ describes a stricter gate: the new model must beat the currently deployed champion by at least 1 accuracy point. Write models/champion.json with {"accuracy": 0.94}, then write a gate that reuses evaluate_model and exits 1 unless the challenger clears that bar. Run it again with the champion at 0.92.

# scripts/gate_vs_champion.py: the challenger must beat the deployed model, not just a floor
import json
import sys
sys.path.insert(0, "scripts")
from evaluate_and_gate import evaluate_model

MIN_GAIN = 0.01   # at least one accuracy point better than the champion

with open("models/champion.json") as f:
    champion = json.load(f)
challenger = evaluate_model("models/model.pkl", "data/training.csv")
gain = challenger["accuracy"] - champion["accuracy"]

print(f"champion accuracy   {champion['accuracy']:.4f}")
print(f"challenger accuracy {challenger['accuracy']:.4f}  (gain {gain:+.4f}, need {MIN_GAIN:+.4f})")
if gain < MIN_GAIN:
    print("BLOCKED: challenger is not clearly better; keep the champion", file=sys.stderr)
    sys.exit(1)
print("PROMOTE: challenger replaces the champion")

$ echo '{"accuracy": 0.94}' > models/champion.json
$ python scripts/gate_vs_champion.py
champion accuracy   0.9400
challenger accuracy 0.9450  (gain +0.0050, need +0.0100)
BLOCKED: challenger is not clearly better; keep the champion
$ echo $?
1
$ echo '{"accuracy": 0.92}' > models/champion.json
$ python scripts/gate_vs_champion.py
champion accuracy   0.9200
challenger accuracy 0.9450  (gain +0.0250, need +0.0100)
PROMOTE: challenger replaces the champion