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.
The differences between a notebook model and a production model at scale are mostly operational, not algorithmic:
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).
# .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"
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
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
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")
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
Based on production experience, here are the failure modes that cost the most time — most of which are not covered in MLOps tutorials:
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.")
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.
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}"
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.
| 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 |
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.