MLOps for 2M+ Data Points: CI/CD, Model Versioning, and What Actually Breaks in Production

MLOps for 2M+ Data Points: CI/CD, Model Versioning, and What Actually Breaks in Production

MLOps looks reasonable in a tutorial. A training pipeline runs, a model gets registered, an API serves predictions. In production with millions of records, weekly retraining cycles, and ten stakeholders who need the model to behave predictably, the interesting problems start. This post covers the specific patterns that work at scale and the failure modes that cost the most time.


What Changes at 2M+ Records

The differences between a notebook model and a production model at scale are mostly operational, not algorithmic:

  • Training time matters. A model that trains in 8 hours cannot be retrained on every code change. You need incremental retraining, caching, and smart test subsets.
  • Data versioning becomes critical. A model trained on data from Q2 is not the same as one trained on Q3 data. Without tracking which dataset produced which model, debugging is nearly impossible.
  • Prediction latency has a p99, not just a mean. The first 1M predictions might be fast; something in the tail of your feature pipeline causes the next 1% to spike to 10x latency. You have to measure p99.
  • Schema changes in source data break feature pipelines silently. An upstream database added a new column, changed a data type, or started producing NULLs in a column that used to be non-null. Your model serves a score, no error is raised, but the feature it depended on is now 0.

CI/CD Pipeline for ML

A CI/CD pipeline for ML has two distinct stages: the code stage (normal software CI) and the model stage (which requires data and has non-deterministic outcomes).

Stage 1: Code CI (every push)

# .github/workflows/ci.yml
name: ML CI

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

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

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: pip install -r requirements.txt -r requirements-dev.txt

      - name: Lint
        run: |
          ruff check src/
          mypy src/ --ignore-missing-imports

      - name: Unit tests (no data required)
        run: pytest tests/unit/ -v --timeout=60

      - name: Feature pipeline tests (sample data)
        run: pytest tests/features/ -v --timeout=120
        env:
          USE_SAMPLE_DATA: "true"

Stage 2: Model CI (on schedule + on demand)

  retrain-and-validate:
    runs-on: self-hosted  # needs GPU / large RAM
    if: github.ref == 'refs/heads/main'
    needs: lint-and-test
    steps:
      - uses: actions/checkout@v4

      - name: Restore feature cache
        uses: actions/cache@v4
        with:
          path: data/features/cache/
          key: features-${{ hashFiles('src/features/**') }}-${{ env.DATA_HASH }}

      - name: Build features
        run: python src/features/build.py --date ${{ env.TRAINING_DATE }}

      - name: Train model
        run: python src/train.py --config configs/prod.yaml

      - name: Validate model
        run: python src/validate.py --min-auc 0.80 --max-drift 0.05

      - name: Register model if validation passes
        run: python src/register.py --registry mlflow --env staging

Validation gate — the step most teams skip

import mlflow
import json
import sys
from pathlib import Path

def validate_model(model_path: str, config: dict) -> bool:
    """
    Validation gate before model registration.
    Returns True if the model passes all checks.
    """
    results = {}

    # 1. Performance check: must beat minimum AUC threshold
    metrics = json.loads(Path(model_path + "/metrics.json").read_text())
    results["auc_ok"] = metrics["val_auc"] >= config["min_auc"]
    results["ap_ok"] = metrics["val_ap"] >= config["min_ap"]

    # 2. Regression check: must not be worse than the current production model
    prod_model = mlflow.pyfunc.load_model("models:/churn-model/Production")
    prod_metrics = json.loads(prod_model.metadata.run_info["metrics"])
    current_auc = prod_metrics.get("val_auc", 0)
    candidate_auc = metrics["val_auc"]
    results["no_regression"] = candidate_auc >= current_auc - config["max_regression_tolerance"]

    # 3. Prediction distribution check: output probabilities should be reasonable
    import pandas as pd
    import numpy as np
    sample = pd.read_parquet("data/validation_sample.parquet")
    candidate = mlflow.pyfunc.load_model(model_path)
    preds = candidate.predict(sample.drop("label", axis=1))
    results["dist_ok"] = (
        preds.mean() > 0.03 and preds.mean() < 0.30 and
        preds.std() > 0.05
    )

    # 4. Feature importance stability: top 5 features shouldn't change drastically
    # (sudden changes in feature importance indicate data issues)
    # ... (implementation specific to your framework)

    all_pass = all(results.values())
    print(f"Validation results: {json.dumps(results, indent=2)}")
    return all_pass


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--model-path", required=True)
    parser.add_argument("--min-auc", type=float, default=0.80)
    parser.add_argument("--min-ap", type=float, default=0.25)
    parser.add_argument("--max-regression-tolerance", type=float, default=0.01)
    args = parser.parse_args()

    passed = validate_model(args.model_path, vars(args))
    sys.exit(0 if passed else 1)  # non-zero exit blocks pipeline progression

Model Versioning with MLflow

import mlflow
import mlflow.lightgbm
from mlflow.tracking import MlflowClient
import lightgbm as lgb

def train_and_register(X_train, y_train, X_val, y_val, config: dict):
    mlflow.set_tracking_uri("http://mlflow-server:5000")
    mlflow.set_experiment("churn-prediction")

    with mlflow.start_run(run_name=f"churn-{config['training_date']}") as run:
        # Log params
        mlflow.log_params({
            "training_date": config["training_date"],
            "n_train": len(X_train),
            "n_val": len(X_val),
            "features": list(X_train.columns),
            **config["model_params"],
        })

        # Log the training data hash for reproducibility
        import hashlib, pandas as pd
        data_hash = hashlib.md5(pd.util.hash_pandas_object(X_train).values).hexdigest()[:8]
        mlflow.set_tag("data_hash", data_hash)
        mlflow.set_tag("git_commit", config.get("git_sha", "unknown"))

        # Train
        dtrain = lgb.Dataset(X_train, label=y_train)
        dval   = lgb.Dataset(X_val, label=y_val, reference=dtrain)
        model  = lgb.train(
            config["model_params"], dtrain,
            num_boost_round=500,
            valid_sets=[dval],
            callbacks=[lgb.early_stopping(50), lgb.log_evaluation(100)],
        )

        # Log metrics
        from sklearn.metrics import roc_auc_score, average_precision_score
        preds = model.predict(X_val)
        metrics = {
            "val_auc": roc_auc_score(y_val, preds),
            "val_ap": average_precision_score(y_val, preds),
        }
        mlflow.log_metrics(metrics)

        # Log model artifact
        mlflow.lightgbm.log_model(
            model,
            artifact_path="model",
            registered_model_name="churn-model",
        )

        return run.info.run_id, metrics


def promote_to_production(run_id: str, model_name: str = "churn-model"):
    """Promote a validated model to Production stage."""
    client = MlflowClient()
    versions = client.search_model_versions(f"name='{model_name}'")
    target = next((v for v in versions if v.run_id == run_id), None)
    if not target:
        raise ValueError(f"No model version found for run {run_id}")

    # Archive current production model
    prod_versions = client.search_model_versions(f"name='{model_name}'")
    for v in prod_versions:
        if v.current_stage == "Production":
            client.transition_model_version_stage(
                name=model_name, version=v.version, stage="Archived"
            )

    # Promote candidate
    client.transition_model_version_stage(
        name=model_name, version=target.version, stage="Production"
    )
    print(f"Promoted {model_name} v{target.version} to Production")

Feature Pipeline Validation — The Silent Killer

The most expensive production incidents in ML systems involve silent feature corruption — the model receives a feature, but the feature value is wrong. No error is raised because the data type is correct; the pipeline just computes garbage.

import great_expectations as gx
from great_expectations.core import ExpectationSuite

def build_feature_expectations(suite_name: str) -> ExpectationSuite:
    """
    Define data quality expectations for the feature matrix.
    Run these before every training and serving job.
    """
    context = gx.get_context()
    suite = context.create_expectation_suite(suite_name, overwrite_existing=True)

    validator = context.get_validator(
        batch_request=...,  # your datasource config
        expectation_suite_name=suite_name,
    )

    # Feature existence checks
    for col in REQUIRED_FEATURES:
        validator.expect_column_to_exist(col)

    # Null rate checks (computed features should rarely be null)
    for col in ["dau_trend", "feature_adoption_pct", "days_to_renewal"]:
        validator.expect_column_values_to_not_be_null(col, mostly=0.95)  # allow max 5% null

    # Value range checks (catch upstream schema changes)
    validator.expect_column_values_to_be_between("feature_adoption_pct", min_value=0.0, max_value=1.0)
    validator.expect_column_values_to_be_between("dau_trend", min_value=-5.0, max_value=5.0)
    validator.expect_column_values_to_be_between("ticket_count_90d", min_value=0, max_value=1000)

    # Distribution checks (catch silent drift in computed features)
    validator.expect_column_mean_to_be_between("feature_adoption_pct", min_value=0.15, max_value=0.85)

    validator.save_expectation_suite(discard_failed_expectations=False)
    return suite


def run_feature_validation(df, suite_name: str) -> bool:
    """Run validation and return True only if all expectations pass."""
    context = gx.get_context()
    results = context.run_validation_operator(
        "action_list_operator",
        assets_to_validate=[...],
    )
    success = results["success"]
    if not success:
        failed = [r for r in results["results"] if not r["success"]]
        print(f"FEATURE VALIDATION FAILED: {len(failed)} checks failed")
        for r in failed[:5]:
            print(f"  {r['expectation_config']['expectation_type']}: {r['result']}")
    return success

What Actually Breaks in Production

Based on production experience, here are the failure modes that cost the most time — most of which are not covered in MLOps tutorials:

1. Time zone bugs in feature engineering

The training data was processed in UTC. The production pipeline runs in the local server timezone. Features computed from timestamps are off by 5-8 hours. Session counts and daily active user metrics shift across day boundaries. The model's feature values in production don't match what it was trained on. AUC in validation: 0.84. Effective AUC in production: 0.67.

Fix: always store and compute timestamps in UTC. Fail loudly if a timestamp column doesn't have timezone info.

def assert_utc(series: pd.Series, col_name: str):
    if not hasattr(series.dtype, 'tz') or series.dtype.tz is None:
        raise ValueError(f"Column '{col_name}' is timezone-naive. Localize to UTC before processing.")
    if str(series.dtype.tz) != "UTC":
        raise ValueError(f"Column '{col_name}' is in {series.dtype.tz}. Convert to UTC first.")

2. Training-serving skew in categorical encoding

A label encoder trained on the training set assigns integer IDs to categories. A new category appears in production that wasn't in training. The encoder raises an error, or silently maps it to an unknown index. Either way, the prediction is wrong.

Fix: use hashing encoders (not label encoders) for high-cardinality categoricals, or always include an "unknown" category in your encoder's vocabulary. Serialize the fitted encoder alongside the model artifact in MLflow.

3. Feature store cache invalidation

Precomputed features are cached daily. The training job uses today's cache. A bug is found and fixed in the feature computation. The cache is not invalidated. The production model is retrained from corrupted cached features. The bug is not discovered until the next audit.

Fix: include a hash of the feature computation code in the cache key.

import hashlib, inspect

def get_feature_cache_key(compute_fn, data_date: str) -> str:
    code_hash = hashlib.md5(inspect.getsource(compute_fn).encode()).hexdigest()[:8]
    return f"features_{data_date}_{code_hash}"

4. Memory leaks in long-running serving processes

A model serving process handles 2M predictions per day. After 72 hours, memory usage has grown from 2GB to 14GB. The process is OOM-killed. Predictions stop. This is almost always caused by accumulating prediction logs, model objects, or intermediate tensors in memory over long runs.

Fix: set memory limits explicitly, implement periodic restarts at low-traffic hours, and use memory profiling tools (tracemalloc, memory-profiler) in pre-production.


Monitoring Checklist for Production ML

Signal What it detects Alert threshold
Input feature null rate (per feature) Upstream schema changes, pipeline failures >5% null for features that are usually non-null
Prediction distribution (mean, std, % in each tier) Feature drift, silent model degradation Mean drifts >0.05 from training baseline
Inference p99 latency Resource contention, memory pressure >200ms for batch, >50ms for online serving
Label feedback rate (when available) Concept drift — ground truth labels are changing Observed churn rate deviates >3% from model-predicted rate
Training job duration Data volume growth, infrastructure issues >2x the median training duration

Summary

MLOps at scale is mostly about discipline: time-zoning all timestamps, versioning data alongside models, validating features before training, and monitoring the right signals in production. The models themselves rarely cause production incidents — it's the surrounding infrastructure that does. Building the CI/CD pipeline, the validation gate, and the monitoring before you need them is the operational difference between a model that runs for two weeks and one that runs reliably for two years.

Related articles