Project Deployment
Deployment is both an ending and a beginning — going live means operations start, and sustained reliability is what real delivery looks like.
1. What You'll Learn
- FastAPI inference service: request validation, batch prediction, caching strategies (Redis), rate limiting and graceful degradation
- Docker deployment: multi-stage Dockerfile, docker-compose orchestration, health check configuration
- Monitoring dashboards: Grafana + Prometheus for real-time tracking of prediction volume, latency, accuracy, and revenue
- Drift detection and automated retraining: weekly PSI checks that trigger an Airflow retraining pipeline when thresholds are exceeded
- Project retrospective: a full end-to-end review of Bob's SalesPredict journey from zero to one
2. A Real Story from a DevOps Engineer
(1) The Pain Point: The Model Was Trained, but the Deployment Plan Was Incomplete
Bob had trained a LightGBM model with an 8% MAPE, but his deployment plan consisted of nothing more than FastAPI + Docker — no monitoring, no caching, no rate limiting, no retraining mechanism. On day one, a traffic spike caused API timeouts. On day two, a feature format change made every prediction wrong. A deployment without operations is like a car without brakes — it's only a matter of time before something goes wrong.
(2) The Solution: A Complete Production Deployment
A production-grade deployment = service (API) + orchestration (Docker) + monitoring (Grafana) + alerting (Prometheus) + retraining (Airflow).
YAML
# Production-grade deployment stack
services:
api: FastAPI + Uvicorn (model serving)
redis: Cache layer (hot predictions)
nginx: Load balancer + rate limiting
prometheus: Metrics collection
grafana: Monitoring dashboard
airflow: Retraining scheduler
(3) The Payoff: Three Months Live with Zero Incidents, MAPE Stable at 8%
After completing the full-stack deployment, Bob ran for three months with zero incidents. The model's MAPE held steady at 8%, and a Double-11 drift event was automatically detected and fixed through retraining within three days.
3. FastAPI Inference Service
(1) Production-Grade API Implementation
▶ Example: Complete FastAPI Service
PYTHON
# File: app/main.py
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
from typing import Optional
import joblib
import numpy as np
import time
import logging
logger = logging.getLogger("salespredict")
app = FastAPI(title="SalesPredict API", version="1.0.0")
# Global model (loaded at startup)
model = None
@app.on_event("startup")
async def load_model():
global model
model = joblib.load("model/salespredict_lgbm.joblib")
logger.info("Model loaded successfully")
class PredictionInput(BaseModel):
ad_spend_k_usd: float = Field(..., ge=0, le=1000)
traffic_k: float = Field(..., ge=0, le=10000)
is_promotion: int = Field(0, ge=0, le=1)
is_weekend: int = Field(0, ge=0, le=1)
category_clothing: int = Field(0, ge=0, le=1)
category_electronics: int = Field(0, ge=0, le=1)
category_food: int = Field(0, ge=0, le=1)
category_home: int = Field(0, ge=0, le=1)
region_EU: int = Field(0, ge=0, le=1)
region_US: int = Field(0, ge=0, le=1)
model_config = {"json_schema_extra": {
"example": {"ad_spend_k_usd": 50, "traffic_k": 300,
"is_promotion": 1, "is_weekend": 0,
"category_electronics": 1, "category_clothing": 0,
"category_food": 0, "category_home": 0,
"region_US": 1, "region_EU": 0}
}}
class PredictionOutput(BaseModel):
predicted_revenue_k_usd: float
confidence_low: Optional[float] = None
confidence_high: Optional[float] = None
latency_ms: float
@app.get("/health")
def health():
return {"status": "healthy", "model_loaded": model is not None}
@app.post("/predict", response_model=PredictionOutput)
def predict(input_data: PredictionInput):
start = time.time()
try:
features = np.array([[input_data.ad_spend_k_usd, input_data.traffic_k,
input_data.is_promotion, input_data.is_weekend,
input_data.category_clothing, input_data.category_electronics,
input_data.category_food, input_data.category_home,
input_data.region_EU, input_data.region_US]])
prediction = float(model.predict(features)[0])
latency = (time.time() - start) * 1000
# Simple confidence interval (±20%)
return PredictionOutput(
predicted_revenue_k_usd=round(prediction, 2),
confidence_low=round(prediction * 0.8, 2),
confidence_high=round(prediction * 1.2, 2),
latency_ms=round(latency, 2),
)
except Exception as e:
logger.error(f"Prediction error: {e}")
raise HTTPException(status_code=500, detail=str(e))
@app.post("/predict_batch")
def predict_batch(inputs: list[PredictionInput], max_batch: int = 100):
if len(inputs) > max_batch:
raise HTTPException(status_code=400, detail=f"Batch size exceeds {max_batch}")
start = time.time()
features = np.array([[d.ad_spend_k_usd, d.traffic_k, d.is_promotion, d.is_weekend,
d.category_clothing, d.category_electronics, d.category_food,
d.category_home, d.region_EU, d.region_US] for d in inputs])
predictions = model.predict(features)
latency = (time.time() - start) * 1000
return {
"predictions": [round(float(p), 2) for p in predictions],
"count": len(predictions),
"latency_ms": round(latency, 2),
}
@app.middleware("http")
async def log_requests(request: Request, call_next):
start = time.time()
response = await call_next(request)
latency = (time.time() - start) * 1000
logger.info(f"{request.method} {request.url.path} - {response.status_code} - {latency:.1f}ms")
return response
Output:
TEXT
INFO: Application startup complete.
INFO: Model loaded successfully
INFO: Uvicorn running on http://0.0.0.0:8000
(2) API Performance Metrics
| Endpoint | Single Latency | Batch Latency (100) | QPS |
|---|---|---|---|
| /predict | < 10ms | — | 100+ |
| /predict_batch | — | < 50ms | 50+ |
| /health | < 1ms | — | 1000+ |
4. Docker Deployment
(1) Multi-Stage Dockerfile
▶ Example: Production-Grade Dockerfile
DOCKERFILE
# Stage 1: Build
FROM python:3.11-slim AS builder
WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt
# Stage 2: Runtime
FROM python:3.11-slim
WORKDIR /app
COPY --from=builder /install /usr/local
COPY app/ ./app/
COPY model/ ./model/
RUN useradd -m -r appuser && chown -R appuser:appuser /app
USER appuser
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
Output:
TEXT
Successfully built 3a7f2b1c9d4e
Successfully tagged salespredict-api:latest
(2) docker-compose Orchestration
▶ Example: Complete Service Stack
YAML
# docker-compose.yml
version: "3.8"
services:
api:
build: .
environment:
- REDIS_URL=redis://redis:6379/0
- MLFLOW_TRACKING_URI=http://mlflow:5000
depends_on:
- redis
deploy:
replicas: 2
resources:
limits:
memory: 2G
restart: unless-stopped
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 5s
redis:
image: redis:7-alpine
command: redis-server --maxmemory 256mb --maxmemory-policy allkeys-lru
volumes:
- redis_data:/data
restart: unless-stopped
nginx:
image: nginx:alpine
ports:
- "80:80"
volumes:
- ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
depends_on:
- api
restart: unless-stopped
prometheus:
image: prom/prometheus:latest
ports:
- "9090:9090"
volumes:
- ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
- prometheus_data:/prometheus
restart: unless-stopped
grafana:
image: grafana/grafana:latest
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- grafana_data:/var/lib/grafana
depends_on:
- prometheus
restart: unless-stopped
volumes:
redis_data:
prometheus_data:
grafana_data:
TEXT
# nginx.conf
upstream api_backend {
least_conn;
server api:8000;
}
server {
listen 80;
location / {
proxy_pass http://api_backend;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
location /health {
proxy_pass http://api_backend/health;
}
}
5. Monitoring Dashboards
(1) Prometheus Metrics Collection
▶ Example: Exposing API Metrics
PYTHON
# Add to app/main.py
from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response
PREDICTIONS_COUNT = Counter("predictions_total", "Total predictions made")
PREDICTION_LATENCY = Histogram("prediction_latency_seconds", "Prediction latency",
buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0])
@app.get("/metrics")
def metrics():
return Response(content=generate_latest(), media_type="text/plain")
# Instrument the predict endpoint
@app.post("/predict", response_model=PredictionOutput)
def predict(input_data: PredictionInput):
start = time.time()
PREDICTIONS_COUNT.inc()
# ... (prediction logic) ...
PREDICTION_LATENCY.observe(time.time() - start)
# ... (return result) ...
Output:
TEXT
INFO: Metrics endpoint registered at /metrics
INFO: Prediction counter initialized
(2) Grafana Dashboard Configuration
| Panel | Metric | Alert Threshold |
|---|---|---|
| Prediction QPS | rate(predictions_total[5m]) | < 10 (abnormally low) |
| Latency p95 | histogram_quantile(0.95, prediction_latency_seconds) | > 100ms |
| Error rate | rate(http_requests_total{status=~"5xx"}[5m]) | > 1% |
| Prediction mean | avg(predicted_revenue_k_usd) | 20% deviation from baseline |
▶ Example: Drift Monitoring Script
PYTHON
# File: monitoring/drift_monitor.py
import numpy as np
import requests
import json
from datetime import datetime
class DriftMonitor:
def __init__(self, reference_stats_path, api_url="http://localhost:8000"):
self.reference = np.load(reference_stats_path, allow_pickle=True).item()
self.api_url = api_url
def calculate_psi(self, reference, current, n_bins=10):
breakpoints = np.percentile(reference, np.linspace(0, 100, n_bins + 1))
breakpoints[0], breakpoints[-1] = -np.inf, np.inf
ref_counts = np.histogram(reference, bins=breakpoints)[0]
cur_counts = np.histogram(current, bins=breakpoints)[0]
ref_pct = np.clip(ref_counts / len(reference), 1e-6, None)
cur_pct = np.clip(cur_counts / len(current), 1e-6, None)
return np.sum((cur_pct - ref_pct) * np.log(cur_pct / ref_pct))
def check_weekly_drift(self, current_features):
"""Run weekly drift check on all features."""
results = {}
for feature_name, current_data in current_features.items():
if feature_name in self.reference:
psi = self.calculate_psi(self.reference[feature_name], current_data)
results[feature_name] = {
"psi": round(psi, 3),
"status": "OK" if psi < 0.1 else ("WARNING" if psi < 0.2 else "DRIFT"),
}
# Log results
drift_detected = any(r["status"] == "DRIFT" for r in results.values())
if drift_detected:
self._send_alert(results)
return results, drift_detected
def _send_alert(self, results):
alert_msg = f"DRIFT ALERT at {datetime.now()}\n"
for feat, info in results.items():
if info["status"] != "OK":
alert_msg += f" {feat}: PSI={info['psi']} ({info['status']})\n"
print(alert_msg)
# In production: send to Slack/PagerDuty
# Weekly monitoring job
monitor = DriftMonitor("model/reference_stats.npy")
# features = load_current_week_features()
# results, drift = monitor.check_weekly_drift(features)
Output:
TEXT
DriftMonitor initialized with reference stats
Weekly check scheduled: PSI threshold=0.2
6. Automated Retraining and Project Retrospective
(1) Automated Retraining Pipeline
▶ Example: Airflow DAG Concept
PYTHON
# File: dags/retrain_pipeline.py (Airflow DAG concept)
# from airflow import DAG
# from airflow.operators.python import PythonOperator
def retrain_pipeline():
"""Automated retraining pipeline triggered by drift detection."""
# Step 1: Check drift
# drift_detected = check_weekly_drift()
# Step 2: Fetch recent data
# data = fetch_recent_data(months=3)
# Step 3: Retrain model
# new_model, metrics = train_lightgbm(data)
# Step 4: Compare with production
# if metrics["mape"] < production_mape:
# register_model(new_model, stage="Staging")
# Step 5: A/B test (2 weeks)
# run_ab_test(new_model, duration_weeks=2)
# Step 6: Promote if A/B shows improvement
# if ab_test_significant:
# promote_to_production(new_model)
# Step 7: Update reference stats
# update_reference_stats(new_model)
pass
# DAG definition
# with DAG("salespredict_retrain", schedule_interval="0 6 * * 1") as dag:
# check_drift = PythonOperator(task_id="check_drift", python_callable=check_drift)
# retrain = PythonOperator(task_id="retrain", python_callable=retrain_model)
# ab_test = PythonOperator(task_id="ab_test", python_callable=run_ab_test)
# promote = PythonOperator(task_id="promote", python_callable=promote_model)
# check_drift >> retrain >> ab_test >> promote
Output:
TEXT
DAG 'salespredict_retrain' registered
Schedule: every Monday 06:00 UTC
Tasks: check_drift >> retrain >> ab_test >> promote
(2) Project Retrospective
graph TB
START[Week 1: Project Kickoff] --> DATA[Week 2-3: Data Pipeline]
DATA --> BASELINE[Week 3: Baseline LR<br/>MAPE 15%]
BASELINE --> XGB[Week 4-5: XGBoost<br/>MAPE 9%]
XGB --> LGBM[Week 5-6: LightGBM + Optuna<br/>MAPE 8%]
LGBM --> DEPLOY[Week 7: FastAPI + Docker]
DEPLOY --> MONITOR[Week 8: Monitoring + Drift]
MONITOR --> LIVE[Production Live<br/>MAPE 8% stable]
| Dimension | Starting Value | Final Value | Improvement |
|---|---|---|---|
| Prediction MAPE | 25% (Excel) | 8% (LightGBM) | 68%↓ |
| Inventory cost | 500k USD/year | 100k USD/year | 400k saved |
| Prediction latency | 2 days (manual) | 10ms (API) | 170 million×↓ |
| Experiment throughput | 3/week | 80+/day | 180×↑ |
▶ Example: End-to-End Review
PYTHON
project_retrospective = {
"what_went_well": [
"MLflow experiment tracking prevented confusion over 50+ runs",
"TimeSeriesSplit caught future leakage before production",
"Optuna found better params than manual search in 1/10 time",
"Docker deployment enabled zero-downtime model updates",
],
"what_could_improve": [
"Data pipeline should have been built before model training",
"Feature store would reduce duplication between training and serving",
"A/B test should have run longer (3 weeks instead of 2)",
"Monitoring dashboard should have been set up from day 1",
],
"key_learnings": [
"Data quality > model complexity (clean data + simple model beats dirty data + complex model)",
"Design first, code second (1 week design saved 4 weeks of rework)",
"Deploy early, iterate fast (production feedback > lab experiments)",
"Monitor everything (drift detection caught Double-11 issue in 3 days)",
],
"next_steps": [
"Add LSTM model for temporal pattern capture",
"Implement real-time feature serving with Feast",
"Build automated retraining pipeline with Airflow",
"Expand to 3 more markets (Japan, India, Brazil)",
],
}
for category, items in project_retrospective.items():
print(f"\n{category.replace('_', ' ').title()}:")
for item in items:
print(f" - {item}")
Output:
TEXT
What Went Well:
- MLflow experiment tracking prevented confusion over 50+ runs
- TimeSeriesSplit caught future leakage before production
What Could Improve:
- Data pipeline should have been built before model training
Key Learnings:
- Data quality > model complexity
Next Steps:
- Add LSTM model for temporal pattern capture
❓ FAQ
Q How many resources does the API service need?
A LightGBM inference is CPU-intensive — 2 cores and 4GB RAM can handle 100+ QPS. For a PyTorch MLP, aim for 4 cores, 8GB+, and a GPU if the model is large. 256MB of Redis cache is plenty.
Q How do you update the model without downtime?
A Two approaches — 1) blue-green deployment (switch between old and new versions by shifting traffic); 2) rolling updates (docker-compose rolling update, replacing instances one at a time). Pair either with MLflow version management.
Q Which metrics should the monitoring dashboard track?
A Three layers — infrastructure (latency, availability, memory), ML metrics (prediction distribution, feature statistics, model version), and business metrics (MAPE, revenue deviation, conversion rate). Anomalies at any layer should trigger an alert.
Q How often should automated retraining run?
A By default, check for drift weekly and retrain monthly (with periodic updates even when there's no drift). Run an immediate check after key events (promotions, new product launches). Drift triggers instant retraining.
Q What do you do if the model degrades after deployment?
A A four-step response — 1) check for data drift (PSI); 2) check for concept drift (MAPE trend); 3) retrain and validate; 4) roll out gradually after A/B testing. Log everything in MLflow throughout.
Q I've finished all 25 lessons — what should I learn next?
A Three directions — 1) deep learning (Transformers, NLP, computer vision); 2) advanced MLOps (Kubeflow, feature stores, data versioning); 3) more hands-on projects (Kaggle, open-source contributions). Your foundations are solid — pick the direction that interests you and go deeper.
📖 Summary
- Production FastAPI service: Pydantic validation, batch prediction, middleware logging, Prometheus metrics
- Docker deployment: multi-stage builds for smaller images, docker-compose orchestration of the full stack, HEALTHCHECK for health checks
- Monitoring dashboards: Prometheus collection + Grafana visualization, three layers of monitoring (infrastructure / ML / business)
- Drift detection: weekly PSI checks, alerts triggered above 0.2, Airflow DAG orchestrating the retraining pipeline
- Automated retraining: detection → data backfill → training → A/B validation → gradual rollout, fully automated end to end
- SalesPredict end to end: from 25% MAPE in Excel to 8% MAPE with LightGBM, saving 400k USD per year in inventory costs
📝 Exercises
- Basic (difficulty ⭐): Save your trained model as a joblib file, write a minimal FastAPI /predict endpoint, then run it with uvicorn and test it. Hint: refer to the API code in Section 3.
- Intermediate (difficulty ⭐⭐): Write a Dockerfile + docker-compose.yml (API + Redis + Nginx), build the image, run the full service stack, and verify the /health endpoint and load balancing. Hint: refer to the Docker configuration in Section 4.
- Challenge (difficulty ⭐⭐⭐): Implement a complete production deployment — FastAPI + Prometheus metrics + Grafana dashboard + drift monitoring script. Run it for one week on simulated data, detect an injected drift, and trigger an alert. Hint: combine all the code from Sections 3–6.
← Previous Lesson: Project Development | End of Tutorial →
