MLOps Implementation: From Notebook to Production

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

Taking a machine learning model from a Jupyter notebook to production is one of the hardest challenges in applied ML. MLOps provides the practices and infrastructure to make this process reliable, reproducible, and scalable.

Infrastructure Setup

Model Serving Options

# Batch predictions (scheduled)
def batch_inference():
    model = load_model("models:/churn-model/Production")
    data = load_new_data()
    predictions = model.predict(data)
    save_predictions(predictions)

# Real-time API (REST)
# FastAPI app (see below)

# Streaming (Kafka + model)
def consume_and_predict():
    consumer = KafkaConsumer('events', bootstrap_servers=['localhost:9092'])
    model = load_model("models:/fraud-model/Production")
    for message in consumer:
        features = extract_features(message.value)
        prediction = model.predict(features)
        producer.send('predictions', prediction)

Docker Containerization

# Dockerfile.ml-service
FROM python:3.10-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    libgomp1 \
    && rm -rf /var/lib/apt/lists/*

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY src/ ./src/
COPY models/ ./models/

# Create non-root user
RUN useradd -m -u 1000 mluser && chown -R mluser:mluser /app
USER mluser

EXPOSE 8000
CMD ["uvicorn", "src.app:app", "--host", "0.0.0.0", "--port", "8000"]

Kubernetes Deployment

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: churn-predictor
  labels:
    app: churn-predictor
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  selector:
    matchLabels:
      app: churn-predictor
  template:
    metadata:
      labels:
        app: churn-predictor
    spec:
      containers:
      - name: predictor
        image: gcr.io/project/churn-predictor:v3.2.1
        ports:
        - containerPort: 8000
        env:
        - name: MODEL_VERSION
          value: "3.2.1"
        - name: MLFLOW_TRACKING_URI
          valueFrom:
            secretKeyRef:
              name: mlflow-secret
              key: tracking-uri
        resources:
          requests:
            memory: "512Mi"
            cpu: "500m"
          limits:
            memory: "1Gi"
            cpu: "1"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: churn-predictor-service
spec:
  selector:
    app: churn-predictor
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP

CI/CD Pipeline Architecture

# .gitlab-ci.yml
stages:
  - data-validation
  - training
  - evaluation
  - staging-deploy
  - integration-test
  - production-deploy

variables:
  MLFLOW_TRACKING_URI: http://mlflow:5000

data-validation:
  stage: data-validation
  script:
    - python -m src.validate_data
    - python -m src.data_quality_report

training:
  stage: training
  script:
    - python -m src.train --experiment churn-model
  artifacts:
    paths:
      - model_artifacts/
    expire_in: 7 days

evaluation:
  stage: evaluation
  script:
    - python -m src.evaluate model_artifacts/
    - python -m src.compare_with_production

staging-deploy:
  stage: staging-deploy
  script:
    - python -m src.deploy --stage staging --model model_artifacts/
  environment:
    name: staging

integration-test:
  stage: integration-test
  script:
    - pytest tests/integration/
    - python -m src.load_test --requests 1000

production-deploy:
  stage: production-deploy
  script:
    - python -m src.deploy --stage production --model model_artifacts/
  environment:
    name: production
  when: manual
  only:
    - main

A/B Testing and Gradual Rollout

# Gradual rollout with traffic splitting
import random

class TrafficSplitter:
    def __init__(self, experiment_name: str, variants: dict):
        """
        variants: {
            'control': {'model': model_v1, 'weight': 0.9},
            'treatment': {'model': model_v2, 'weight': 0.1}
        }
        """
        self.experiment_name = experiment_name
        self.variants = variants
        self._validate_weights()

    def select_variant(self, user_id: str) -> str:
        """Consistent assignment based on user ID."""
        hash_val = hash(f"{self.experiment_name}:{user_id}") % 10000
        cumulative = 0
        for variant, config in self.variants.items():
            cumulative += config['weight'] * 10000
            if hash_val < cumulative:
                return variant
        return list(self.variants.keys())[-1]

    def predict(self, features, user_id):
        variant = self.select_variant(user_id)
        model = self.variants[variant]['model']
        prediction = model.predict(features)

        # Log for analysis
        self._log_prediction(user_id, variant, prediction)
        return prediction

Monitoring Stack

# Prometheus metrics for ML
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from fastapi import Response
import time

# Define metrics
PREDICTIONS = Counter('model_predictions_total', 'Total predictions', ['model_version'])
PREDICTION_LATENCY = Histogram('model_prediction_seconds', 'Prediction latency', ['model_version'])
PREDICTION_SCORE = Gauge('model_prediction_score', 'Current prediction score')
DRIFT_SCORE = Gauge('model_drift_score', 'Data drift score', ['feature'])

@app.post("/predict")
async def predict(request: PredictionRequest):
    start = time.time()

    # Predict
    prediction = model.predict([request.features])[0]
    latency = time.time() - start

    # Update metrics
    PREDICTIONS.labels(model_version=MODEL_VERSION).inc()
    PREDICTION_LATENCY.labels(model_version=MODEL_VERSION).observe(latency)
    PREDICTION_SCORE.set(prediction)

    return {"prediction": prediction}

@app.get("/metrics")
async def metrics():
    return Response(content=generate_latest(), media_type="text/plain")

Shadow Deployment

Deploy a new model in “shadow mode” — it receives traffic but its predictions are not served to users:

class ShadowDeployer:
    def __init__(self, primary_model, shadow_model):
        self.primary = primary_model
        self.shadow = shadow_model

    def predict(self, features):
        # Primary model serves the user
        primary_pred = self.primary.predict(features)

        # Shadow model runs in background for comparison
        shadow_pred = self.shadow.predict(features)

        # Log comparison for analysis
        self._log_comparison(features, primary_pred, shadow_pred)

        # Only return primary prediction
        return primary_pred

    def _log_comparison(self, features, primary, shadow):
        """Store comparison data for offline evaluation."""
        log_entry = {
            'timestamp': datetime.now(),
            'features': features,
            'primary_prediction': primary,
            'shadow_prediction': shadow,
            'match': primary == shadow
        }
        # Store for batch analysis
        self.analysis_buffer.append(log_entry)

Data Quality Monitoring

from great_expectations.dataset import PandasDataset
import great_expectations as ge

def validate_incoming_data(df: pd.DataFrame) -> bool:
    """Validate data quality before inference."""
    dataset = PandasDataset(df)

    expectations = [
        dataset.expect_column_values_to_not_be_null('customer_id'),
        dataset.expect_column_values_to_be_between(
            'total_purchases', min_value=0, max_value=10000
        ),
        dataset.expect_column_values_to_be_in_set(
            'plan_type', ['basic', 'premium', 'enterprise']
        ),
        dataset.expect_column_mean_to_be_between(
            'avg_order_value', min_value=0, max_value=5000
        ),
    ]

    for expectation in expectations:
        if not expectation.success:
            logging.warning(f"Data validation failed: {expectation.expectation_config}")

    return all(e.success for e in expectations)

Rollback Strategy

class ModelRouter:
    def __init__(self):
        self.active_models = {
            'production': load_model('models:/churn-model/Production'),
            'rollback': load_model('models:/churn-model/Production_previous'),
            'canary': None,
        }
        self.current_version = 'production'
        self.monitor = ModelMonitor()

    def predict(self, features):
        try:
            prediction = self.active_models[self.current_version].predict(features)
            self.monitor.record_success()
            return prediction
        except Exception as e:
            self.monitor.record_failure()
            if self.monitor.error_rate > 0.1:  # 10% error rate threshold
                self.rollback()
            return self.active_models['rollback'].predict(features)

    def rollback(self):
        logging.warning(f"Rolling back from {self.current_version}")
        self.current_version = 'rollback'
        notify_team(f"Automated rollback to previous model version")

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: How do I version ML models in production? A: Use semantic versioning (v3.2.1) with a model registry (MLflow). Tag each version with its training run ID, data version, and performance metrics.

Q: How do I test models in production safely? A: Use shadow deployment (run new model without serving), A/B testing (small traffic percentage), or canary deployments (gradual rollout with monitoring).

Q: What monitoring is essential for production ML? A: Prediction distribution (drift), latency/throughput, error rate, feature distributions, and when available, accuracy on delayed ground truth.

Q: How often should I retrain? A: Schedule-based (weekly/monthly) for stable environments. Trigger-based (upon detected drift or accuracy drop) for dynamic environments. Always validate new models before deployment.

Q: How do I handle data drift? A: Monitor feature distributions with statistical tests (PSI, KS-test, Mahalanobis distance). Set thresholds that trigger alerts and optional automatic retraining.

Q: What is the minimum viable monitoring for a single model? A: Log predictions and features, track distribution shifts, monitor latency and error rates, and set up alerts for anomalies. Start simple and add sophistication as needed.

Production ML Architecture

Feature Store Integration

# Feast feature store example
from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="feature_repo/")

# Get training data
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_features:avg_session_duration",
        "user_features:purchase_count_7d",
        "item_features:item_popularity_score",
        "context_features:hour_of_day",
    ]
).to_df()

# Get online features for inference
feature_vector = store.get_online_features(
    features=[
        "user_features:avg_session_duration",
        "item_features:item_popularity_score",
    ],
    entity_rows=[{"user_id": "abc123", "item_id": "xyz789"}]
).to_dict()

A/B Testing Framework

# Serve multiple model versions simultaneously
class ModelRouter:
    def __init__(self):
        self.models = {
            "v1": load_model("model_v1.pkl"),
            "v2": load_model("model_v2.pkl"),
        }
        self.traffic_split = {"v1": 0.8, "v2": 0.2}
    
    def predict(self, features, user_id):
        # Deterministic routing based on user_id hash
        bucket = hash(user_id) % 100
        if bucket < self.traffic_split["v1"] * 100:
            return self.models["v1"].predict(features)
        else:
            return self.models["v2"].predict(features)

Model Monitoring

# Track prediction distribution drift
from scipy.stats import ks_2samp

def detect_drift(reference_scores, current_scores, threshold=0.05):
    statistic, p_value = ks_2samp(reference_scores, current_scores)
    drift_detected = p_value < threshold
    return drift_detected, {"ks_statistic": statistic, "p_value": p_value}