Model Deployment¶

Model deployment is getting ML models into production. This sheet covers serving strategies, APIs, batch inference, containerization, and deployment patterns.

Table of Contents¶

Installation / Setup
Core Syntax
Common Operations
Practical Examples
Common Patterns
Best Practices
Common Mistakes
Troubleshooting
Related Cheat Sheets
Official Documentation

Installation / Setup¶

pip install fastapi uvicorn mlflow
pip install boto3  # for S3

Core Syntax¶

Deployment Strategies¶

Strategy	Description	When to Use
REST API	Synchronous HTTP endpoint	Low-latency, real-time
Batch	Process data in batches	High-throughput, non-real-time
Streaming	Process data as it arrives	Real-time data pipelines
Edge	Deploy to edge devices	Mobile, IoT
Serverless	Deploy as functions	Variable load, cost optimization

Serving Patterns¶

# FastAPI (recommended for Python)
from fastapi import FastAPI
import mlflow

app = FastAPI()
model = mlflow.pyfunc.load_model("models:/my_model/production")

@app.post("/predict")
def predict(data: dict):
    return {"prediction": model.predict(data)}

Common Operations¶

FastAPI Serving¶

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import mlflow
import pandas as pd

app = FastAPI()

# Load model at startup
model = mlflow.pyfunc.load_model("models:/my_model/production")

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

class PredictResponse(BaseModel):
    predictions: list[float]

@app.post("/predict", response_model=PredictResponse)
def predict(request: PredictRequest):
    try:
        df = pd.DataFrame(request.features, columns=["f1", "f2", "f3"])
        predictions = model.predict(df).tolist()
        return PredictResponse(predictions=predictions)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

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

Batch Inference¶

import mlflow
import pandas as pd

# Load model
model = mlflow.pyfunc.load_model("models:/my_model/production")

# Read input data
df = pd.read_parquet("data/to_predict.parquet")

# Predict in batches
batch_size = 1000
predictions = []
for i in range(0, len(df), batch_size):
    batch = df.iloc[i:i+batch_size]
    pred = model.predict(batch)
    predictions.extend(pred)

# Save results
df["prediction"] = predictions
df.to_parquet("data/predictions.parquet")

Docker Container¶

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt

COPY . .
EXPOSE 8000

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# main.py
from fastapi import FastAPI
import mlflow

app = FastAPI()
model = mlflow.pyfunc.load_model("models:/my_model/production")

@app.post("/predict")
def predict(data: dict):
    return {"prediction": model.predict(data)}

MLflow Serving¶

# Serve model
mlflow models serve -m "models:/my_model/production" -p 5001

# Predict
curl -X POST http://localhost:5001/invocations \
  -H "Content-Type: application/json" \
  -d '{"inputs": [[1.0, 2.0, 3.0]]}'

Model Export Formats¶

# PyTorch → ONNX
import torch
torch.onnx.export(model, dummy_input, "model.onnx")

# scikit-learn → PMML
from sklearn2pmml import sklearn2pmml
sklearn2pmml(model, "model.pmml")

# TensorFlow → SavedModel
model.save("model_savedmodel")

# PyTorch → TorchScript
scripted = torch.jit.script(model)
scripted.save("model.pt")

Practical Examples¶

Example 1 — Production API¶

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import mlflow
import pandas as pd
import logging

app = FastAPI(title="Churn Prediction API")
logger = logging.getLogger(__name__)

# Load model
model = mlflow.pyfunc.load_model("models:/churn_model/production")

class CustomerData(BaseModel):
    age: float
    income: float
    tenure_months: int
    contract_type: str

class PredictionResponse(BaseModel):
    churn_probability: float
    will_churn: bool

@app.post("/predict", response_model=PredictionResponse)
def predict(customer: CustomerData):
    try:
        df = pd.DataFrame([customer.dict()])
        proba = model.predict_proba(df)[0][1]
        return PredictionResponse(
            churn_probability=proba,
            will_churn=proba > 0.5
        )
    except Exception as e:
        logger.error(f"Prediction failed: {e}")
        raise HTTPException(status_code=500, detail=str(e))

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

Example 2 — Batch pipeline¶

import mlflow
import pandas as pd
from datetime import datetime

def run_batch_prediction():
    # Load model
    model = mlflow.pyfunc.load_model("models:/churn_model/production")

    # Read data
    df = pd.read_parquet("s3://data/customers/current/")

    # Predict
    features = ["age", "income", "tenure_months", "contract_type"]
    df["churn_probability"] = model.predict_proba(df[features])[:, 1]
    df["will_churn"] = df["churn_probability"] > 0.5

    # Save
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    df.to_parquet(f"s3://data/predictions/{timestamp}.parquet")

    print(f"Processed {len(df)} customers")

if __name__ == "__main__":
    run_batch_prediction()

Example 3 — Docker deployment¶

# docker-compose.yml
version: '3.8'
services:
  model-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - MLFLOW_TRACKING_URI=http://mlflow:5000
    depends_on:
      - mlflow

  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.10.0
    ports:
      - "5000:5000"
    command: mlflow server --host 0.0.0.0 --port 5000

Common Patterns¶

Pattern: A/B testing¶

import random

def get_model_version(user_id):
    # Deterministic routing based on user_id
    hash_val = hash(user_id) % 100
    if hash_val < 10:  # 10% traffic to new model
        return "new_model"
    return "old_model"

@app.post("/predict")
def predict(data: dict, user_id: str):
    model_version = get_model_version(user_id)
    model = load_model(model_version)
    return {"prediction": model.predict(data), "model": model_version}

Pattern: canary deployment¶

import random

canary_percentage = 0.05  # 5% traffic

@app.post("/predict")
def predict(data: dict):
    if random.random() < canary_percentage:
        return new_model.predict(data)
    return old_model.predict(data)

Pattern: rollback¶

def rollback():
    client = mlflow.tracking.MlflowClient()
    versions = client.get_latest_versions("churn_model", stages=["production"])
    if len(versions) > 1:
        # Rollback to previous version
        client.transition_model_version_stage(
            "churn_model", versions[1].version, "production"
        )
        print(f"Rolled back to version {versions[1].version}")

Pattern: model caching¶

from functools import lru_cache
import mlflow

@lru_cache(maxsize=1)
def get_model():
    return mlflow.pyfunc.load_model("models:/my_model/production")

@app.post("/predict")
def predict(data: dict):
    model = get_model()
    return {"prediction": model.predict(data)}

Best Practices¶

Use health checks — always include /health endpoint.
Version your API — /v1/predict, /v2/predict.
Validate inputs — use Pydantic models.
Add logging — track requests and errors.
Use timeouts — prevent hanging requests.
Containerize — Docker for reproducibility.
Monitor latency — track response times.
Implement rollback — keep previous versions ready.

Common Mistakes¶

Mistake	Fix
No health check	Add `/health` endpoint
No input validation	Use Pydantic models
No logging	Add request/error logging
No timeouts	Set request timeouts
No rollback strategy	Keep previous versions
Hardcoding model paths	Use model registry

Troubleshooting¶

Symptom	Cause / Fix
Slow predictions	Profile model; batch predictions
Out of memory	Reduce batch size; use GPU
Model not found	Check model path; check registry
High latency	Use async; cache model
Deployment fails	Check dependencies; check Docker

mlops.md — MLOps overview
mlflow.md — model registry
model-monitoring.md — production monitoring
fastapi — FastAPI serving
docker — containerization
docker-compose — multi-service deployment

Official Documentation¶

FastAPI — https://fastapi.tiangolo.com/
MLflow Serving — https://mlflow.org/docs/latest/models.html#deploying-models
Docker — https://docs.docker.com/

Last Updated¶

2026-09-06 — Initial version.

Model Deployment¶

Table of Contents¶

Installation / Setup¶

Core Syntax¶

Deployment Strategies¶

Serving Patterns¶

Common Operations¶

FastAPI Serving¶

Batch Inference¶

Docker Container¶

MLflow Serving¶

Model Export Formats¶

Practical Examples¶

Example 1 — Production API¶

Example 2 — Batch pipeline¶

Example 3 — Docker deployment¶

Common Patterns¶

Pattern: A/B testing¶

Pattern: canary deployment¶

Pattern: rollback¶

Pattern: model caching¶

Best Practices¶

Common Mistakes¶

Troubleshooting¶

Related Cheat Sheets¶

Official Documentation¶

Last Updated¶