MLOps Guide: Machine Learning Operations

Machine Learning Machine Learning 7 min read 1422 words Beginner ExcellentWiki Editorial Team

MLOps (Machine Learning Operations) applies DevOps principles to machine learning systems. It bridges the gap between model development and production deployment, addressing the unique challenges of ML: data dependencies, model versioning, reproducibility, monitoring, and continuous retraining.

The ML Lifecycle

Unlike traditional software, ML systems have additional components:

Data collection and validation — ensuring data quality, schema consistency
Feature engineering and storage — transforming raw data into features
Experiment tracking — recording hyperparameters, metrics, model versions
Model training and validation — reproducible training pipelines
Model deployment — serving infrastructure (REST, batch, edge)
Monitoring and observability — data drift, model decay, performance
Retraining — automated or triggered by monitoring signals
Governance and compliance — lineage, fairness, explainability

Experiment Tracking

Track experiments systematically to avoid configuration chaos:

import mlflow

# Set tracking URI
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("customer-churn-v2")

with mlflow.start_run() as run:
    # Log parameters
    mlflow.log_param("model_type", "xgboost")
    mlflow.log_param("n_estimators", 500)
    mlflow.log_param("learning_rate", 0.05)
    mlflow.log_param("max_depth", 6)

    # Log metrics
    mlflow.log_metric("accuracy", 0.923)
    mlflow.log_metric("f1_score", 0.891)
    mlflow.log_metric("auc_roc", 0.952)

    # Log model
    mlflow.sklearn.log_model(model, "model")

    # Log artifacts (feature importance plots, confusion matrix)
    mlflow.log_artifact("feature_importance.png")
    mlflow.log_artifact("confusion_matrix.png")

    # Log tags for searchability
    mlflow.set_tag("dataset", "churn_2024_v3")
    mlflow.set_tag("team", "customer-analytics")

Experiment Tracking Tools

Tool	Hosting	Strengths
MLflow	Self-hosted, Databricks	Industry standard, full lifecycle
Weights & Biases	Cloud, self-hosted	Rich visualization, collaboration
Neptune.ai	Cloud	Team features, comparison UI
DVC	Self-hosted	Git-based, open source

Model Versioning and Registry

import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient()

# Register a model
mlflow.register_model(
    "runs:/<run_id>/model",
    "customer-churn-predictor"
)

# Promote model to stages
client.transition_model_version_stage(
    name="customer-churn-predictor",
    version=3,
    stage="Staging"          # None, Staging, Production, Archived
)

# Get production model
production_model = mlflow.pyfunc.load_model(
    "models:/customer-churn-predictor/Production"
)

Pipeline Orchestration

ML Pipelines with Kubeflow

import kfp
from kfp import dsl

@dsl.pipeline(
    name="Churn Prediction Pipeline",
    description="End-to-end ML pipeline for customer churn"
)
def churn_pipeline(
    data_path: str,
    model_type: str = "xgboost",
    test_size: float = 0.2,
):
    ingest_op = dsl.ContainerOp(
        name="ingest",
        image="gcr.io/project/data-ingest:latest",
        arguments=["--data-path", data_path]
    )

    validate_op = dsl.ContainerOp(
        name="validate",
        image="gcr.io/project/data-validate:latest",
        arguments=["--input", ingest_op.output]
    )

    train_op = dsl.ContainerOp(
        name="train",
        image="gcr.io/project/model-train:latest",
        arguments=[
            "--data", validate_op.output,
            "--model-type", model_type,
            "--test-size", str(test_size),
        ]
    )

    evaluate_op = dsl.ContainerOp(
        name="evaluate",
        image="gcr.io/project/model-evaluate:latest",
        arguments=["--model", train_op.output]
    )

    deploy_op = dsl.ContainerOp(
        name="deploy",
        image="gcr.io/project/model-deploy:latest",
        arguments=["--model", train_op.output],
        # Only deploy if evaluation passes threshold
    ).after(evaluate_op)

Orchestration Tools

Kubeflow — Kubernetes-native ML platform
Apache Airflow — General-purpose DAG orchestration
Prefect — Modern Python workflow engine
ZenML — ML-specific pipeline framework
TFX — TensorFlow Extended (Google’s production pipeline)

Feature Store

A centralized feature store ensures consistency between training and serving:

# Feast Feature Store
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")

# Get training features
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "customer_features:total_purchases",
        "customer_features:avg_order_value",
        "customer_features:days_since_last_purchase",
        "transaction_features:rolling_30d_amount",
    ]
).to_df()

# Get serving features (online)
feature_vector = store.get_online_features(
    features=[
        "customer_features:total_purchases",
        "customer_features:avg_order_value",
    ],
    entity_rows=[{"customer_id": "abc123"}]
).to_dict()

Model Deployment

REST API with FastAPI

from fastapi import FastAPI
from pydantic import BaseModel
import mlflow
import numpy as np

app = FastAPI(title="Churn Prediction API")

# Load model at startup
model = mlflow.pyfunc.load_model("models:/customer-churn-predictor/Production")

class PredictionRequest(BaseModel):
    features: list[float]

class PredictionResponse(BaseModel):
    churn_probability: float
    prediction: int

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    features = np.array(request.features).reshape(1, -1)
    probability = model.predict_proba(features)[0, 1]
    prediction = int(probability >= 0.5)

    return PredictionResponse(
        churn_probability=float(probability),
        prediction=prediction
    )

@app.get("/health")
async def health():
    return {"status": "healthy"}

Batch Prediction

def batch_predict(input_path: str, output_path: str):
    """Generate predictions for large datasets."""
    model = mlflow.pyfunc.load_model("models:/churn-model/Production")
    data = pd.read_parquet(input_path)

    # Process in chunks for memory efficiency
    predictions = []
    for chunk in np.array_split(data, 100):
        chunk_preds = model.predict(chunk)
        predictions.extend(chunk_preds)

    # Save results
    data["prediction"] = predictions
    data.to_parquet(output_path)

Monitoring

Monitor both model performance and data characteristics:

# Data drift detection with Evidently
from evidently.metrics import DataDriftTable
from evidently.report import Report

data_drift_report = Report(metrics=[DataDriftTable()])
data_drift_report.run(
    reference_data=reference_data,
    current_data=current_data
)
data_drift_report.save_html("data_drift.html")

# Performance monitoring
def log_prediction(features, prediction, actual=None):
    """Log each prediction for monitoring."""
    monitoring_log = {
        "timestamp": datetime.now().isoformat(),
        "features": features,
        "prediction": prediction,
        "actual": actual,
        "model_version": "v3.2.1"
    }
    # Send to monitoring system (e.g., Prometheus, Datadog)
    send_metric("model_predictions", 1)
    send_metric("prediction_score", prediction)

    if actual is not None:
        accuracy = 1.0 if prediction == actual else 0.0
        send_metric("model_accuracy", accuracy)

Key Monitoring Metrics

Prediction distribution — has the output distribution shifted?
Feature distribution — data drift on input features
Model accuracy — when ground truth is available
Latency and throughput — serving infrastructure performance
Error rate — prediction failures or timeouts

CI/CD for ML

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

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

jobs:
  train-and-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.10'

      - name: Validate data
        run: python -m src.validate_data

      - name: Train model
        run: python -m src.train_model

      - name: Evaluate model
        run: python -m src.evaluate_model

      - name: Register model (if performance improves)
        run: python -m src.register_model

      - name: Deploy to staging
        run: python -m src.deploy_model --stage staging

      - name: Run integration tests
        run: pytest tests/integration/

      - name: Promote to production
        run: python -m src.promote_to_production

Real-World Implementation Tips

Production Considerations

When moving from development to production, several factors become critical. Error handling should be comprehensive — every external call (database, API, file system) should have proper error checking, logging, and retry logic where appropriate. Performance monitoring through metrics and structured logging helps identify bottlenecks before they affect users.

Testing Strategy

A thorough testing approach combines multiple levels:

Unit tests verify individual functions and methods in isolation
Integration tests validate that components work together correctly
Edge case tests cover boundary conditions, empty inputs, and error states
Performance tests ensure the system meets latency and throughput requirements

Test data should be realistic but controlled. Mock external dependencies to make tests fast and deterministic. Aim for tests that are independent, repeatable, and fast enough to run on every commit.

Documentation

Good documentation is essential for maintainable code. Follow these principles:

Document the “why” not just the “what” — explain design decisions
Keep examples up to date with the code
Include usage examples for public APIs
Document configuration options and their defaults
Explain error conditions and recovery strategies

Security Best Practices

Security should be considered throughout development:

Validate all inputs at system boundaries
Use parameterized queries for database access
Store secrets in environment variables or secret managers
Keep dependencies updated to patch vulnerabilities
Apply the principle of least privilege

Performance Optimization

Optimize based on measured data, not assumptions:

Profile before optimizing — identify actual bottlenecks
Measure the impact of each change
Consider the trade-off between speed and readability
Cache expensive operations with appropriate invalidation
Use connection pooling for database and network resources

Monitoring and Observability

Production systems need visibility:

Structured logging with correlation IDs for request tracking
Metrics for latency, throughput, error rates, and resource usage
Health check endpoints for load balancers and orchestration
Distributed tracing for request flows across services
Alerts for anomaly detection based on baselines

These patterns apply across all programming languages and frameworks. The specific implementation varies, but the principles remain consistent.

FAQ

Q: What is the difference between MLOps and DevOps? A: MLOps extends DevOps with ML-specific concerns: experiment tracking, data/feature versioning, model registry, data drift monitoring, and retraining pipelines.

Q: Do I need MLOps for small projects? A: Basic practices (experiment tracking, version control for code and data, reproducibility) benefit any project. Full MLOps infrastructure is proportional to team size and model count.

Q: How do I handle model retraining? A: Retrain on a schedule (weekly/monthly) or trigger retraining when monitoring detects drift, performance degradation, or new labeled data becomes available.

Q: What is a feature store? A: A feature store is a centralized repository for feature definitions and values. It ensures consistency between training and serving, avoids duplicate feature engineering, and provides both batch (historical) and online (real-time) access.

Q: How do I version ML models? A: Use a model registry (MLflow Model Registry, S3 with versioning) with semantic versioning. Store the model artifact alongside its metadata: training code version, data version, hyperparameters, and performance metrics.

Q: What is the minimum viable MLOps setup? A: (1) Code version control (git), (2) experiment tracking (MLflow), (3) model registry, (4) basic monitoring (prediction distribution shifts), (5) reproducible training pipeline.

MLOps Implementation Details

Experiment Tracking with MLflow

import mlflow
import mlflow.sklearn

mlflow.set_experiment("customer-churn-v2")

with mlflow.start_run():
    params = {
        "n_estimators": 200,
        "max_depth": 8,
        "learning_rate": 0.1,
        "subsample": 0.8
    }
    mlflow.log_params(params)
    
    model = GradientBoostingClassifier(**params)
    model.fit(X_train, y_train)
    
    accuracy = model.score(X_test, y_test)
    mlflow.log_metric("accuracy", accuracy)
    mlflow.sklearn.log_model(model, "model")
    
    # Log artifacts
    mlflow.log_artifact("feature_importance.png")
    mlflow.log_artifact("confusion_matrix.png")

Model Registry

# Register model via MLflow CLI
mlflow models register -m "runs/<run_id>/model" -n "churn-model"

# Promote model to staging/production
mlflow models transition -m "models:/churn-model/5" -s "Production"

Data and Model Versioning

import dvc.api

# Track data with DVC
# dvc add data/training_data.csv
# git add data/training_data.csv.dvc

# Load specific version
url = "https://github.com/user/repo/raw/main/data/training_data.csv"
repo = "/path/to/repo"
data = dvc.api.read(url, repo=repo, rev="v2.1.0")