Model Serving and Deployment
Deploy trained ML models to production — REST APIs with FastAPI, BentoML, batch inference, and model versioning strategies.
Real-World Scenario
A data science team trains a churn prediction model weekly. The first version is a pickle file emailed around. By month three, nobody knows which model is running in production. Model serving infrastructure solves this: each model version is registered, validated, deployed to staging, then promoted to production with a single command.
Serving a Model with FastAPI
# serve.py — production-ready model serving with FastAPI
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, field_validator
from pathlib import Path
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="Churn Prediction API", version="1.0.0")
# Load model at startup (not per-request)
MODEL_PATH = Path("./models/churn_v2.pkl")
model = None
@app.on_event("startup")
async def load_model():
global model
if not MODEL_PATH.exists():
raise RuntimeError(f"Model not found: {MODEL_PATH}")
model = joblib.load(MODEL_PATH)
logger.info(f"Loaded model from {MODEL_PATH}")
class PredictionRequest(BaseModel):
customer_id: str
tenure_months: int
monthly_charges: float
total_charges: float
contract_type: str # "Month-to-month", "One year", "Two year"
has_internet: bool
support_calls_30d: int
@field_validator("tenure_months")
@classmethod
def check_tenure(cls, v):
if v < 0 or v > 600:
raise ValueError("tenure_months must be 0-600")
return v
@field_validator("monthly_charges")
@classmethod
def check_charges(cls, v):
if v < 0 or v > 10_000:
raise ValueError("monthly_charges must be non-negative")
return v
class PredictionResponse(BaseModel):
customer_id: str
churn_prob: float
churn_label: bool
risk_tier: str # LOW / MEDIUM / HIGH
latency_ms: float
model_version: str
def engineer_features(req: PredictionRequest) -> np.ndarray:
contract_map = {"Month-to-month": 0, "One year": 1, "Two year": 2}
return np.array([[
req.tenure_months,
req.monthly_charges,
req.total_charges,
contract_map.get(req.contract_type, 0),
int(req.has_internet),
req.support_calls_30d,
req.monthly_charges / max(req.tenure_months, 1), # avg monthly spend
]])
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
if model is None:
raise HTTPException(status_code=503, detail="Model not loaded")
t0 = time.perf_counter()
try:
features = engineer_features(request)
prob = float(model.predict_proba(features)[0, 1])
label = prob >= 0.5
risk_tier = "HIGH" if prob >= 0.7 else ("MEDIUM" if prob >= 0.4 else "LOW")
except Exception as e:
logger.error(f"Prediction failed for {request.customer_id}: {e}")
raise HTTPException(status_code=500, detail="Prediction failed")
latency = (time.perf_counter() - t0) * 1000
logger.info(f"customer={request.customer_id} prob={prob:.3f} latency={latency:.1f}ms")
return PredictionResponse(
customer_id=request.customer_id,
churn_prob=round(prob, 4),
churn_label=label,
risk_tier=risk_tier,
latency_ms=round(latency, 2),
model_version="2.0.0",
)
@app.get("/health")
async def health():
return {"status": "ok", "model_loaded": model is not None}
# To run: uvicorn serve:app --host 0.0.0.0 --port 8000 --workers 4Batch Inference Pipeline
# batch_inference.py — process large datasets offline
import joblib
import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime
import logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)
CHUNK_SIZE = 10_000 # process in chunks to avoid OOM on large files
MODEL_PATH = Path("./models/churn_v2.pkl")
INPUT_PATH = Path("./data/customers_to_score.parquet")
OUTPUT_PATH = Path("./output/churn_scores_{date}.parquet")
def engineer_features_batch(df: pd.DataFrame) -> np.ndarray:
contract_map = {"Month-to-month": 0, "One year": 1, "Two year": 2}
return np.column_stack([
df["tenure_months"].fillna(0),
df["monthly_charges"].fillna(0),
df["total_charges"].fillna(0),
df["contract_type"].map(contract_map).fillna(0),
df["has_internet"].astype(int),
df["support_calls_30d"].fillna(0),
(df["monthly_charges"] / df["tenure_months"].clip(lower=1)).fillna(0),
])
def run_batch_inference(
model_path: Path,
input_path: Path,
output_path: Path,
chunk_size: int = CHUNK_SIZE,
) -> dict:
model = joblib.load(model_path)
logger.info(f"Loaded model: {model_path}")
total_rows = 0
results = []
date_str = datetime.now().strftime("%Y%m%d_%H%M")
logger.info(f"Processing {input_path} in chunks of {chunk_size}")
for i, chunk in enumerate(pd.read_parquet(input_path, chunksize=chunk_size)):
t0 = datetime.now()
features = engineer_features_batch(chunk)
probs = model.predict_proba(features)[:, 1]
chunk_results = pd.DataFrame({
"customer_id": chunk["customer_id"],
"churn_prob": np.round(probs, 4),
"churn_label": probs >= 0.5,
"risk_tier": pd.cut(probs,
bins=[-0.001, 0.4, 0.7, 1.001],
labels=["LOW", "MEDIUM", "HIGH"]),
"scored_at": datetime.now().isoformat(),
"model_version": "2.0.0",
})
results.append(chunk_results)
total_rows += len(chunk)
elapsed = (datetime.now() - t0).total_seconds()
logger.info(f"Chunk {i+1}: {len(chunk)} rows in {elapsed:.2f}s "
f"({len(chunk)/elapsed:.0f} rows/s)")
output_df = pd.concat(results, ignore_index=True)
final_path = Path(str(output_path).format(date=date_str))
output_df.to_parquet(final_path, index=False)
stats = {
"total_rows": total_rows,
"high_risk": int((output_df["risk_tier"] == "HIGH").sum()),
"medium_risk": int((output_df["risk_tier"] == "MEDIUM").sum()),
"low_risk": int((output_df["risk_tier"] == "LOW").sum()),
"output_path": str(final_path),
}
logger.info(f"Done: {stats}")
return stats
if __name__ == "__main__":
run_batch_inference(MODEL_PATH, INPUT_PATH, OUTPUT_PATH)Blue-Green Deployment with Model Versioning
# deployment.py — safe model promotion strategy
import mlflow
import requests
import subprocess
import sys
MLFLOW_URI = "http://localhost:5000"
mlflow.set_tracking_uri(MLFLOW_URI)
client = mlflow.tracking.MlflowClient()
def get_production_model(model_name: str) -> dict | None:
"""Return the current production model version and metadata."""
try:
versions = client.get_latest_versions(model_name, stages=["Production"])
if not versions:
return None
v = versions[0]
return {"version": v.version, "run_id": v.run_id, "status": v.current_stage}
except Exception:
return None
def run_smoke_tests(endpoint_url: str, threshold: float = 0.95) -> bool:
"""Run smoke tests against a staging endpoint."""
test_cases = [
{
"payload": {
"customer_id": "smoke_1",
"tenure_months": 24,
"monthly_charges": 65.0,
"total_charges": 1560.0,
"contract_type": "One year",
"has_internet": True,
"support_calls_30d": 1,
},
"expected_risk": ["LOW", "MEDIUM"],
},
{
"payload": {
"customer_id": "smoke_2",
"tenure_months": 1,
"monthly_charges": 95.0,
"total_charges": 95.0,
"contract_type": "Month-to-month",
"has_internet": True,
"support_calls_30d": 8,
},
"expected_risk": ["HIGH", "MEDIUM"],
},
]
passed = 0
for case in test_cases:
try:
r = requests.post(f"{endpoint_url}/predict", json=case["payload"], timeout=5)
if r.status_code == 200:
risk = r.json()["risk_tier"]
if risk in case["expected_risk"]:
passed += 1
continue
print(f"FAIL: status={r.status_code}, risk={r.json().get('risk_tier')}")
except Exception as e:
print(f"ERROR: {e}")
pass_rate = passed / len(test_cases)
print(f"Smoke tests: {passed}/{len(test_cases)} passed ({pass_rate:.0%})")
return pass_rate >= threshold
def promote_model(model_name: str, version: str) -> bool:
"""Promote a model version to production after staging validation."""
# Step 1: confirm it passes in staging
staging_url = "http://staging-model-server:8001"
if not run_smoke_tests(staging_url):
print("Smoke tests failed — aborting promotion")
return False
# Step 2: archive the current production model
current = get_production_model(model_name)
if current:
client.transition_model_version_stage(
name=model_name,
version=current["version"],
stage="Archived",
)
print(f"Archived v{current['version']}")
# Step 3: promote the new version
client.transition_model_version_stage(
name=model_name,
version=version,
stage="Production",
)
print(f"Promoted v{version} to Production")
return True
# Example: promote the latest staging model
if __name__ == "__main__":
MODEL_NAME = "churn_classifier"
staging_versions = client.get_latest_versions(MODEL_NAME, stages=["Staging"])
if staging_versions:
v = staging_versions[0].version
success = promote_model(MODEL_NAME, v)
sys.exit(0 if success else 1)
else:
print("No staging model to promote")
sys.exit(1)Frequently Asked Questions
What is the difference between online serving and batch inference?
Online serving responds to individual requests in real time (low latency, typically <100ms). Batch inference processes large datasets offline on a schedule — predictions are pre-computed and stored. Use online for user-facing features (recommendations, fraud detection at checkout), batch for nightly reports, bulk scoring, or retraining pipelines.
How do I handle model versioning in production?
Store model artifacts with a version identifier in a model registry (MLflow, SageMaker, or a simple S3 path with semver). Your serving layer loads a specific version by name. Use a staging → production promotion workflow: deploy to staging, run smoke tests, then promote. Never deploy a new model directly to production without testing on real traffic first.