MLOps basics: concepts, trade-offs and production lessons

MLOps basics: concepts, trade-offs and production lessons

Most teams train models in notebooks, push them to production by hand, and wonder why inference accuracy tanks three weeks later. MLOps basics: concepts, trade-offs and production lessons brings the discipline of software engineering to machine learning—automating everything from data preparation through monitoring, so your models stay reliable in production.

Machine learning broke the traditional deployment model. A trained model artifact is inert until it meets live data, which shifts constantly. Features drift, data distributions change, and silent failures become the norm. We've learned from production disasters in e-commerce recommendation systems, fraud detection pipelines, and demand forecasting that treating ML like software—with versioning, testing, and observability—is non-negotiable.

This post covers the four pillars of MLOps: automated pipelines, model CI, registry and deployment patterns, and monitoring feedback loops. We'll focus on the concrete trade-offs and what actually works at scale.

Pipelines — Automate data prep, training and evaluation

A pipeline orchestrates the journey from raw data to a trained, evaluated model. Without automation, you're re-running scripts manually, losing reproducibility, and introducing human error. With a pipeline, every experiment follows the same path, every artifact is versioned, and you can trace what data went into which model.

Tools like Airflow, Prefect, or Kubeflow handle orchestration. For smaller teams, cloud-native options (AWS SageMaker Pipelines, Vertex AI) reduce operational overhead. The trade-off: managed services cost more but eliminate infrastructure toil; open-source tools are cheaper but demand DevOps expertise.

A typical pipeline has these stages:

• Data ingestion — Pull from data warehouse, apply access controls, log source version.
• Validation — Check schema, detect null rates, verify row counts haven't collapsed.
• Feature engineering — Transform raw columns into model inputs; cache intermediate results to avoid recomputation.
• Train/test split — Respect temporal order for time-series; stratify for imbalanced classification.
• Training — Fit the model, serialize weights and hyperparameters.
• Evaluation — Compute metrics on held-out test data; compare against baseline.
• Promotion decision — Auto-promote if metrics pass thresholds, else alert and block.

Here's a simplified Airflow DAG that ties these together:

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'ml-team',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'fraud_model_pipeline',
    default_args=default_args,
    schedule_interval='0 2 * * *',  # 2 AM daily
    catchup=False,
)

def fetch_data(**context):
    # Query warehouse for last 30 days of transactions
    df = query_warehouse("SELECT * FROM transactions WHERE created_at > now() - interval '30 days'")
    context['task_instance'].xcom_push(key='raw_data_path', value='/tmp/raw.parquet')
    df.to_parquet('/tmp/raw.parquet')

def validate_data(**context):
    df = pd.read_parquet('/tmp/raw.parquet')
    assert df.shape[0] > 100000, "Data too small"
    assert df.isnull().sum().max() < 0.5, "Nulls exceed threshold"
    print(f"Validation passed: {df.shape}")

def engineer_features(**context):
    df = pd.read_parquet('/tmp/raw.parquet')
    df['amount_log'] = np.log1p(df['amount'])
    df['hour'] = pd.to_datetime(df['timestamp']).dt.hour
    df = df.drop(['timestamp'], axis=1)
    df.to_parquet('/tmp/features.parquet')

def train_model(**context):
    X = pd.read_parquet('/tmp/features.parquet')
    y = X.pop('is_fraud')
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
    
    model = XGBClassifier(max_depth=5, learning_rate=0.1, n_estimators=100)
    model.fit(X_train, y_train, eval_set=[(X_test, y_test)], early_stopping_rounds=10)
    
    model.save_model('/tmp/model.pkl')
    context['task_instance'].xcom_push(key='model_path', value='/tmp/model.pkl')

def evaluate_model(**context):
    model = XGBClassifier()
    model.load_model('/tmp/model.pkl')
    X_test = pd.read_parquet('/tmp/features.parquet').iloc[-100000:]
    y_test = X_test.pop('is_fraud')
    
    auc = roc_auc_score(y_test, model.predict_proba(X_test)[:, 1])
    precision = precision_score(y_test, model.predict(X_test))
    
    print(f"AUC: {auc:.4f}, Precision: {precision:.4f}")
    assert auc > 0.80, f"AUC {auc} below threshold"
    
    context['task_instance'].xcom_push(key='metrics', value={'auc': auc, 'precision': precision})

t1 = PythonOperator(task_id='fetch_data', python_callable=fetch_data, dag=dag)
t2 = PythonOperator(task_id='validate_data', python_callable=validate_data, dag=dag)
t3 = PythonOperator(task_id='engineer_features', python_callable=engineer_features, dag=dag)
t4 = PythonOperator(task_id='train_model', python_callable=train_model, dag=dag)
t5 = PythonOperator(task_id='evaluate_model', python_callable=evaluate_model, dag=dag)

t1 >> t2 >> t3 >> t4 >> t5

This DAG runs nightly. Each step is independent and retryable. If validation fails, the pipeline stops early and alerts the team. If training succeeds but evaluation flunks the AUC check, the old model stays in production.

CI for models — Run tests and metric checks on every change

Traditional software CI runs unit tests; model CI runs metric tests. When a teammate changes feature engineering logic or hyperparameters, CI should train a candidate model, evaluate it against the baseline, and reject the PR if metrics regress.

The workflow:

1. Developer pushes a branch with feature or model code changes.
2. CI pipeline trains a candidate model on the same data as the baseline.
3. Compare candidate metrics to baseline using a statistical test (not just point estimates, which are noisy).
4. Block merge if candidate underperforms by more than a threshold (e.g., AUC drop > 0.02).
5. Approve merge and promote candidate to staging if metrics are better or within tolerance.

Set up a GitHub Actions workflow (or equivalent) that runs on every PR:

name: Model CI

on: [pull_request]

jobs:
  test-model:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
        with:
          fetch-depth: 0  # Fetch full history to diff feature code
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Download baseline metrics
        run: aws s3 cp s3://ml-artifacts/baseline_metrics.json baseline.json
      
      - name: Train candidate model
        run: python train.py --config config.yaml --output /tmp/candidate_model.pkl
      
      - name: Evaluate candidate
        run: python evaluate.py --model /tmp/candidate_model.pkl --output /tmp/candidate_metrics.json
      
      - name: Compare metrics
        run: |
          python scripts/compare_metrics.py \
            --baseline baseline.json \
            --candidate /tmp/candidate_metrics.json \
            --threshold 0.02
      
      - name: Upload candidate artifacts on success
        if: success()
        run: |
          aws s3 cp /tmp/candidate_model.pkl s3://ml-artifacts/pr-${{ github.event.pull_request.number }}/model.pkl
          aws s3 cp /tmp/candidate_metrics.json s3://ml-artifacts/pr-${{ github.event.pull_request.number }}/metrics.json

The comparison script uses statistical rigor. Don't just compare point estimates; use bootstrapped confidence intervals or permutation tests to assess whether differences are meaningful.

def compare_metrics(baseline, candidate, threshold=0.02):
    """Compare candidate to baseline using 95% confidence intervals."""
    for metric in ['auc', 'precision', 'recall']:
        base_val = baseline[metric]
        cand_val = candidate[metric]
        diff = cand_val - base_val
        
        if diff < -threshold:
            raise AssertionError(
                f"{metric} regressed: baseline {base_val:.4f}, candidate {cand_val:.4f}, "
                f"diff {diff:.4f} exceeds threshold {-threshold:.4f}"
            )
        else:
            print(f"✓ {metric}: {cand_val:.4f} (delta: {diff:+.4f})")

Trade-off: Model CI is expensive—training takes minutes to hours. You'll pay for compute on every PR. Budget for it in your cloud spend, or gate CI to PRs from trusted reviewers only.

Registry and deployment — Promote models through staging to production

A model registry is a versioned artifact store with metadata. It tracks which model is in staging, which is in production, and who approved each promotion. Tools: MLflow Model Registry, AWS SageMaker Model Registry, Hugging Face Hub, or a custom S3 + DynamoDB solution.

The promotion flow:

Development → Staging → Production

Each stage runs in isolation. Staging uses production-like data volumes and latencies but serves internal dashboards. Production serves real customer requests.

Store model metadata alongside the artifact:

import mlflow
import json

# Log model during training
mlflow.set_experiment('fraud-detection')
mlflow.start_run()
mlflow.log_params({
    'max_depth': 5,
    'learning_rate': 0.1,
    'n_estimators': 100,
})
mlflow.log_metrics({
    'auc': 0.845,
    'precision': 0.88,
    'recall': 0.79,
})
mlflow.sklearn.log_model(model, 'model', registered_model_name='fraud-detector')
mlflow.end_run()

# Query registry to get latest production model
client = mlflow.tracking.MlflowClient()
model_version = client.get_latest_versions(name='fraud-detector', stages=['Production'])[0]
print(f"Production model: {model_version.source}")  # S3 path or artifact URI

# Register candidate in Staging
client.transition_model_version_stage(
    name='fraud-detector',
    version=model_version.version + 1,
    stage='Staging',
)

# After 24h validation, promote to Production
client.transition_model_version_stage(
    name='fraud-detector',
    version=model_version.version + 1,
    stage='Production',
)

At deployment time, your serving infrastructure (FastAPI, Lambda, Sagemaker endpoint) pulls the Production model version from the registry and loads it. A deployment coordinator service (or manual promotion for safety) triggers the cutover.

Trade-off: Blue-green deployment (run old and new model in parallel before switching traffic) gives safety but doubles compute cost. Canary deployments (route 5% of traffic to new model) reduce blast radius but complicate monitoring. For mission-critical systems (fraud, compliance), we prefer staged rollouts; for less critical ones (personalization), faster rollouts are fine.

Monitoring — Close the loop with drift and performance checks

The moment your model hits production, it begins to decay. Data drifts away from training distribution, feature pipelines break, and metrics silently degrade. Monitoring closes the feedback loop: you detect problems, retrain, and promote a new model.

Monitor three things:

1. Prediction drift — Model outputs shift (e.g., fraud scores drop). Indicates feature or data change.
2. Data drift — Input features shift (e.g., average transaction amount rises). Retraining may help.
3. Performance drift — True metrics degrade (e.g., AUC falls). Usually detected via ground-truth labels when available with delay.

Implement drift detection in your serving layer:

from scipy.stats import ks_2samp
import time

class DriftDetector:
    def __init__(self, baseline_data, threshold=0.05):
        self.baseline = baseline_data  # Features from training
        self.threshold = threshold
        self.current_window = []
        self.window_size = 10000
    
    def check_drift(self, feature_vector):
        self.current_window.append(feature_vector)
        
        if len(self.current_window) >= self.window_size:
            # Compare current window to baseline
            for feature_name in feature_vector.keys():
                baseline_vals = self.baseline[feature_name]
                current_vals = [row[feature_name] for row in self.current_window]
                
                # Kolmogorov-Smirnov test for distribution shift
                statistic, p_value = ks_2samp(baseline_vals, current_vals)
                
                if p_value < self.threshold:
                    print(f"DRIFT ALERT: {feature_name} p={p_value:.4f}")
                    self.alert_team(feature_name, statistic, p_value)
            
            self.current_window = []  # Reset window

detector = DriftDetector(baseline_training_data)

@app.post('/predict')
def predict(features: dict):
    detector.check_drift(features)
    prediction = model.predict([features])[0]
    return {'prediction': prediction}

Log predictions and outcomes (once ground truth arrives) to evaluate true performance:

import logging

logger = logging.getLogger(__name__)

@app.post('/predict')
def predict(features: dict):
    prediction = model.predict([features])[0]
    request_id = str(uuid.uuid4())
    
    # Log for monitoring
    logger.info(json.dumps({
        'request_id': request_id,
        'timestamp': datetime.utcnow().isoformat(),
        'features': features,
        'prediction': float(prediction),
    }))
    
    return {'prediction': prediction, 'request_id': request_id}

@app.post('/feedback')
def feedback(request_id: str, label: bool):
    """Called later when ground truth is known (e.g., transaction confirmed fraudulent)."""
    logger.info(json.dumps({
        'request_id': request_id,
        'label': label,
        'timestamp': datetime.utcnow().isoformat(),
    }))
    # Queries over these logs compute daily/weekly AUC, precision, recall
    return {'status': 'recorded'}

Aggregate logs into a monitoring dashboard. Alert if drift or performance crosses thresholds. Trigger automatic retraining (or manual review for high-stakes systems) when metrics degrade.

Common mistakes

• Training-serving skew: Features computed differently in pipeline vs. serving code. Use shared feature libraries or centralized feature stores (Feast, Tecton) to avoid this trap.
• Leaking test data into training. If you use the same data for hyperparameter tuning and evaluation, metrics are optimistically biased. Use separate test sets or nested cross-validation.
• Ignoring class imbalance in data drift detection. A skew-sensitive metric like AUC can mask catastrophic failures on minority classes. Monitor per-class metrics.
• Deploying without monitoring. Many teams train carefully but go dark in production. Set up monitoring first; deploy second.
• Manual model promotion. Copy-pasting model files to production is error-prone. Automate promotion through a registry; gate it with approval workflows for high-risk models.
• Offline-only evaluation. A model that beats baseline offline may fail in production due to distribution shift or feedback loops. Always run a staging canary before full rollout.

Key takeaways

• Automate data prep, training, and evaluation with an orchestrated pipeline so every model is reproducible and traced back to its data sources.
• Run model CI on every code change—train a candidate, compare metrics to baseline, and block PRs that regress performance.
• Use a model registry to version artifacts and enforce staged promotion from development through staging to production, avoiding ad-hoc deployments.
• Close the feedback loop with monitoring: detect data and prediction drift, track true performance metrics as ground truth arrives, and trigger retraining when degradation crosses thresholds.
• Invest in feature engineering consistency (shared libraries, feature stores) to avoid training-serving skew, a silent killer in production systems.
• Treat model deployment as seriously as application deployment—approval workflows, canary rollouts, and easy rollback are essential for production reliability.