Master the complete MLOps lifecycle: experiment tracking, model versioning, containerization with Docker, orchestration with Kubernetes, model serving APIs, monitoring, and production deployment best practices.
Machine Learning Operations (MLOps) bridges the gap between ML development and production deployment. While building accurate models is important, deploying them to production where they serve millions of users reliably is equally critical.
In October 2025, MLOps has matured into a essential discipline combining DevOps practices with ML-specific challenges: model versioning, experiment tracking, data drift detection, online learning, and A/B testing. Companies like Netflix, Uber, Airbnb, and Google rely on robust MLOps practices to deploy thousands of models serving billions of predictions daily.
This final lesson covers the complete MLOps lifecycle—taking your AI models from Jupyter notebooks to production systems that are monitored, maintained, and continuously improved. You'll learn industry-standard tools (MLflow, Docker, Kubernetes, FastAPI), best practices, and real-world deployment patterns used by leading tech companies.
By the end, you'll have the skills to deploy and maintain production ML systems that deliver business value at scale.
MLOps is a set of practices to deploy and maintain ML models in production reliably and efficiently.
Traditional Software:
1Code → Build → Test → Deploy → Monitor
2 ↑ │
3 └──────── Bug fixes ────────┘ML Systems (more complex):
1Data → Feature Engineering → Model Training → Evaluation
2 ↓ ↓ ↓ ↓
3 │ │ │ │
4 └──────────────┴──────────────────┴─────────────┘
5 │
6 Model Registry
7 │
8 ┌────────┴────────┐
9 │ │
10 Serving Monitoring
11 │ │
12 └────────┬────────┘
13 │
14 Retrain trigger (data drift, performance decay)
15 │
16 └───────► Back to Data1. Experimentation Phase
1# Data scientist workflow
2- Explore data (EDA)
3- Feature engineering
4- Model selection and tuning
5- Track experiments (MLflow, W&B)
6- Version control (Git, DVC)
7
8Goal: Find best model for the task2. Model Development
1# Best practices
2- Reproducible environments (Docker)
3- Automated testing (pytest)
4- Code quality (linting, type hints)
5- Documentation (model cards)
6
7Goal: Production-ready model code3. Model Packaging
1# Serialization formats
2- Pickle (Python objects)
3- ONNX (cross-framework)
4- SavedModel (TensorFlow)
5- TorchScript (PyTorch)
6
7# Include:
8- Model weights
9- Preprocessing pipeline
10- Dependencies (requirements.txt)
11- Inference code
12
13Goal: Portable, reproducible model artifact4. Model Serving
1# Deployment patterns
2- REST API (FastAPI, Flask)
3- Batch prediction (Airflow, Luigi)
4- Streaming (Kafka, Flink)
5- Edge deployment (TensorFlow Lite, ONNX Runtime)
6
7Goal: Expose model for predictions5. Monitoring & Maintenance
1# What to monitor
2- Model performance (accuracy, latency)
3- Data drift (input distribution shift)
4- Concept drift (target distribution shift)
5- System health (CPU, memory, errors)
6
7Goal: Detect issues, trigger retraining1. Reproducibility
1Problem: "It works on my machine"
2Solution:
3- Pin dependencies (requirements.txt)
4- Use containers (Docker)
5- Version data (DVC, S3 versioning)
6- Seed random number generators
7- Track experiments (MLflow)2. Automation
1Manual processes → Automated pipelines
2
3- CI/CD: Automated testing and deployment
4- Scheduled retraining: Airflow DAGs
5- Monitoring alerts: Prometheus + Grafana
6- A/B testing: Automated traffic splitting3. Versioning Everything
1Code: Git
2Data: DVC, S3 versioning
3Models: MLflow Model Registry
4Configs: Git, ConfigMaps (Kubernetes)
5Infrastructure: Terraform (IaC)4. Scalability
1Design for scale from day 1:
2- Horizontal scaling (Kubernetes)
3- Load balancing (nginx, AWS ALB)
4- Caching (Redis)
5- Async processing (Celery)
6- GPU utilization (batch inference)5. Security
1- Authentication (API keys, OAuth)
2- Authorization (RBAC)
3- Encryption (TLS, data-at-rest)
4- Input validation (prevent injection)
5- Rate limiting (prevent abuse)
6- Audit logging (track access)Level 0: No MLOps
Level 1: DevOps, No MLOps
Level 2: Automated Training
Level 3: Automated Deployment
Level 4: Full MLOps
Most companies in October 2025 aim for Level 3, with Level 4 reserved for cutting-edge organizations.
Experiment tracking records ML experiments for reproducibility and comparison. Model registry manages production model versions.
Problem without tracking:
1data_scientist_1.ipynb
2data_scientist_1_final.ipynb
3data_scientist_1_final_v2.ipynb
4data_scientist_1_final_v2_REALLY_FINAL.ipynb
5
6Which model did we deploy? What hyperparameters?
7Lost in the notebook graveyard! 💀Solution: Structured experiment tracking
1import mlflow
2import mlflow.sklearn
3from sklearn.ensemble import RandomForestClassifier
4from sklearn.metrics import accuracy_score
5
6# Start MLflow run
7with mlflow.start_run(run_name="rf_experiment_1"):
8 # Log parameters
9 mlflow.log_param("n_estimators", 100)
10 mlflow.log_param("max_depth", 10)
11 mlflow.log_param("random_state", 42)
12
13 # Train model
14 model = RandomForestClassifier(
15 n_estimators=100,
16 max_depth=10,
17 random_state=42
18 )
19 model.fit(X_train, y_train)
20
21 # Evaluate
22 y_pred = model.predict(X_test)
23 accuracy = accuracy_score(y_test, y_pred)
24
25 # Log metrics
26 mlflow.log_metric("accuracy", accuracy)
27 mlflow.log_metric("test_samples", len(X_test))
28
29 # Log model
30 mlflow.sklearn.log_model(model, "random_forest_model")
31
32 # Log artifacts (plots, data)
33 plt.figure()
34 plot_confusion_matrix(y_test, y_pred)
35 plt.savefig("confusion_matrix.png")
36 mlflow.log_artifact("confusion_matrix.png")
37
38 print(f"Run ID: {mlflow.active_run().info.run_id}")MLflow UI: Browse experiments, compare runs, visualize metrics
1# Start MLflow UI
2mlflow ui --port 5000
3
4# Navigate to http://localhost:5000
5# Features:
6# - Compare multiple runs
7# - Visualize metrics over time
8# - Download models and artifacts
9# - Filter and search experimentsPurpose: Manage model versions and deployment stages
1# Register model
2model_uri = f"runs:/{run_id}/random_forest_model"
3mlflow.register_model(model_uri, "ChurnPredictionModel")
4
5# Transition model through stages
6from mlflow.tracking import MlflowClient
7
8client = MlflowClient()
9
10# Stage 1: Staging
11client.transition_model_version_stage(
12 name="ChurnPredictionModel",
13 version=1,
14 stage="Staging"
15)
16
17# Run tests on staging model
18# If tests pass:
19
20# Stage 2: Production
21client.transition_model_version_stage(
22 name="ChurnPredictionModel",
23 version=1,
24 stage="Production"
25)
26
27# Load production model
28model = mlflow.pyfunc.load_model(
29 "models:/ChurnPredictionModel/Production"
30)
31
32# Make predictions
33predictions = model.predict(new_data)1# Search experiments
2from mlflow.tracking import MlflowClient
3
4client = MlflowClient()
5experiment = client.get_experiment_by_name("churn_prediction")
6
7# Get all runs
8runs = client.search_runs(
9 experiment_ids=[experiment.experiment_id],
10 order_by=["metrics.accuracy DESC"],
11 max_results=10
12)
13
14# Compare runs
15import pandas as pd
16
17data = []
18for run in runs:
19 data.append({
20 'run_id': run.info.run_id,
21 'n_estimators': run.data.params.get('n_estimators'),
22 'max_depth': run.data.params.get('max_depth'),
23 'accuracy': run.data.metrics.get('accuracy'),
24 'f1_score': run.data.metrics.get('f1_score')
25 })
26
27df = pd.DataFrame(data)
28print(df.sort_values('accuracy', ascending=False))
29
30# Output:
31# run_id n_estimators max_depth accuracy f1_score
32# 0 abc123 200 15 0.924 0.918
33# 1 def456 100 10 0.918 0.912
34# 2 ghi789 50 8 0.903 0.897Weights & Biases (W&B)
1import wandb
2
3# Initialize
4wandb.init(project="churn_prediction", name="exp_1")
5
6# Log everything
7wandb.config.update({
8 "n_estimators": 100,
9 "max_depth": 10
10})
11
12# During training
13for epoch in range(n_epochs):
14 loss = train_step()
15 wandb.log({"loss": loss, "epoch": epoch})
16
17# Log model
18wandb.save("model.h5")
19
20# Features:
21# - Real-time dashboard
22# - Hyperparameter sweeps
23# - Team collaboration
24# - Report generationDVC (Data Version Control)
1# Track large files (data, models)
2dvc add data/train.csv
3git add data/train.csv.dvc
4git commit -m "Add training data"
5
6# Push to remote storage (S3, GCS, Azure)
7dvc push
8
9# On another machine
10git pull
11dvc pull # Downloads data from remote
12
13# Benefits:
14# - Git-like versioning for data
15# - Supports S3, GCS, Azure, HDFS
16# - Lightweight (Git tracks .dvc files, not data)Model serving exposes trained models via APIs for real-time predictions.
FastAPI is the modern choice for ML model serving (October 2025).
1# model_server.py
2from fastapi import FastAPI, HTTPException
3from pydantic import BaseModel
4import joblib
5import numpy as np
6
7# Load model at startup
8model = joblib.load("models/churn_model.pkl")
9
10app = FastAPI(
11 title="Churn Prediction API",
12 description="Predict customer churn probability",
13 version="1.0.0"
14)
15
16# Request schema
17class PredictionRequest(BaseModel):
18 tenure: int
19 monthly_charges: float
20 total_charges: float
21 contract_type: str
22 internet_service: str
23
24 class Config:
25 json_schema_extra = {
26 "example": {
27 "tenure": 12,
28 "monthly_charges": 70.5,
29 "total_charges": 846.0,
30 "contract_type": "Month-to-month",
31 "internet_service": "Fiber optic"
32 }
33 }
34
35# Response schema
36class PredictionResponse(BaseModel):
37 churn_probability: float
38 prediction: str
39 confidence: float
40
41# Health check
42@app.get("/health")
43async def health():
44 """Check if service is healthy"""
45 return {
46 "status": "healthy",
47 "model": "churn_model_v1",
48 "version": "1.0.0"
49 }
50
51# Prediction endpoint
52@app.post("/predict", response_model=PredictionResponse)
53async def predict(request: PredictionRequest):
54 """
55 Predict churn probability for a customer
56
57 Returns probability between 0 and 1:
58 - < 0.3: Low risk
59 - 0.3-0.7: Medium risk
60 - > 0.7: High risk
61 """
62 try:
63 # Prepare features
64 features = np.array([[
65 request.tenure,
66 request.monthly_charges,
67 request.total_charges,
68 1 if request.contract_type == "Month-to-month" else 0,
69 1 if request.internet_service == "Fiber optic" else 0
70 ]])
71
72 # Predict
73 proba = model.predict_proba(features)[0]
74 churn_prob = float(proba[1])
75
76 # Classify
77 if churn_prob < 0.3:
78 prediction = "Low Risk"
79 elif churn_prob < 0.7:
80 prediction = "Medium Risk"
81 else:
82 prediction = "High Risk"
83
84 return PredictionResponse(
85 churn_probability=churn_prob,
86 prediction=prediction,
87 confidence=max(proba)
88 )
89
90 except Exception as e:
91 raise HTTPException(
92 status_code=500,
93 detail=f"Prediction failed: {str(e)}"
94 )
95
96# Batch prediction
97@app.post("/predict/batch")
98async def predict_batch(requests: list[PredictionRequest]):
99 """Predict for multiple customers at once"""
100 results = []
101 for req in requests:
102 pred = await predict(req)
103 results.append(pred)
104 return results
105
106# Model info
107@app.get("/model/info")
108async def model_info():
109 """Get model metadata"""
110 return {
111 "model_type": "RandomForestClassifier",
112 "n_features": 5,
113 "feature_names": [
114 "tenure",
115 "monthly_charges",
116 "total_charges",
117 "contract_month_to_month",
118 "internet_fiber"
119 ],
120 "trained_date": "2025-10-01",
121 "accuracy": 0.924,
122 "f1_score": 0.918
123 }
124
125# Run server
126if __name__ == "__main__":
127 import uvicorn
128 uvicorn.run(app, host="0.0.0.0", port=8000)Running and testing:
1# Start server
2python model_server.py
3
4# Test with curl
5curl -X POST "http://localhost:8000/predict" \
6 -H "Content-Type: application/json" \
7 -d '{
8 "tenure": 12,
9 "monthly_charges": 70.5,
10 "total_charges": 846.0,
11 "contract_type": "Month-to-month",
12 "internet_service": "Fiber optic"
13 }'
14
15# Response:
16{
17 "churn_probability": 0.73,
18 "prediction": "High Risk",
19 "confidence": 0.73
20}
21
22# Interactive docs (Swagger UI)
23# Open: http://localhost:8000/docs1. Request Validation
1from pydantic import BaseModel, Field, validator
2
3class PredictionRequest(BaseModel):
4 tenure: int = Field(ge=0, le=100) # 0 ≤ tenure ≤ 100
5 monthly_charges: float = Field(gt=0) # > 0
6
7 @validator('tenure')
8 def validate_tenure(cls, v):
9 if v < 0:
10 raise ValueError('Tenure must be non-negative')
11 return v2. Rate Limiting
1from slowapi import Limiter, _rate_limit_exceeded_handler
2from slowapi.util import get_remote_address
3
4limiter = Limiter(key_func=get_remote_address)
5app.state.limiter = limiter
6
7@app.post("/predict")
8@limiter.limit("100/minute") # Max 100 requests per minute
9async def predict(request: PredictionRequest):
10 # ... prediction code3. Authentication
1from fastapi import Depends, HTTPException, status
2from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
3
4security = HTTPBearer()
5
6def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
7 """Verify API key"""
8 if credentials.credentials != "your_secret_api_key":
9 raise HTTPException(
10 status_code=status.HTTP_401_UNAUTHORIZED,
11 detail="Invalid API key"
12 )
13 return credentials.credentials
14
15@app.post("/predict")
16async def predict(
17 request: PredictionRequest,
18 token: str = Depends(verify_token)
19):
20 # Only accessible with valid API key
21 # ... prediction code4. Logging and Monitoring
1import logging
2from datetime import datetime
3
4logging.basicConfig(level=logging.INFO)
5logger = logging.getLogger(__name__)
6
7@app.post("/predict")
8async def predict(request: PredictionRequest):
9 start_time = datetime.now()
10
11 # Log request
12 logger.info(f"Prediction request received: {request.dict()}")
13
14 # Make prediction
15 result = await make_prediction(request)
16
17 # Log latency
18 latency = (datetime.now() - start_time).total_seconds()
19 logger.info(f"Prediction completed in {latency:.3f}s")
20
21 return resultFor TensorFlow models, use TensorFlow Serving:
1# Export model
2model.save("saved_model/1")
3
4# Run TensorFlow Serving (Docker)
5docker run -p 8501:8501 \
6 --mount type=bind,source=$(pwd)/saved_model,target=/models/my_model \
7 -e MODEL_NAME=my_model \
8 -t tensorflow/serving
9
10# Make predictions
11curl -X POST http://localhost:8501/v1/models/my_model:predict \
12 -d '{"instances": [[1.0, 2.0, 3.0]]}'1# Create model archive
2torch-model-archiver \
3 --model-name churn_model \
4 --version 1.0 \
5 --model-file model.py \
6 --serialized-file model.pth \
7 --handler custom_handler.py
8
9# Start TorchServe
10torchserve --start \
11 --model-store model_store \
12 --models churn_model=churn_model.mar
13
14# Predict
15curl http://localhost:8080/predictions/churn_model \
16 -F "[email protected]"1. Online Serving (Real-time)
2. Batch Serving (Offline)
3. Streaming
Docker packages your model with all dependencies into a portable container.
Problem: "Works on my machine"
1Data Scientist:
2- Python 3.10, TensorFlow 2.14, CUDA 11.8
3- macOS
4
5Production:
6- Python 3.8, TensorFlow 2.10, CUDA 11.2
7- Ubuntu 20.04
8
9Result: Model doesn't work! 😱Solution: Docker
1Container includes:
2- Exact Python version
3- All dependencies (requirements.txt)
4- System libraries
5- Configuration
6
7Result: Works everywhere! 🎉1# Dockerfile
2# Multi-stage build for smaller image
3
4# Stage 1: Build dependencies
5FROM python:3.10-slim as builder
6
7WORKDIR /build
8
9# Install build tools
10RUN apt-get update && apt-get install -y \
11 build-essential \
12 gcc \
13 && rm -rf /var/lib/apt/lists/*
14
15# Copy and install Python dependencies
16COPY requirements.txt .
17RUN pip install --no-cache-dir --user -r requirements.txt
18
19# Stage 2: Runtime
20FROM python:3.10-slim
21
22WORKDIR /app
23
24# Copy Python packages from builder
25COPY /root/.local /root/.local
26
27# Make sure scripts in .local are usable
28ENV PATH=/root/.local/bin:$PATH
29
30# Copy application code
31COPY app/ ./app/
32COPY models/ ./models/
33COPY model_server.py .
34
35# Create non-root user
36RUN useradd -m -u 1000 modeluser && \
37 chown -R modeluser:modeluser /app
38USER modeluser
39
40# Expose port
41EXPOSE 8000
42
43# Health check
44HEALTHCHECK \
45 CMD curl -f http://localhost:8000/health || exit 1
46
47# Run server
48CMD ["uvicorn", "model_server:app", "--host", "0.0.0.0", "--port", "8000"]requirements.txt:
1fastapi==0.104.1
2uvicorn[standard]==0.24.0
3pydantic==2.5.0
4numpy==1.24.3
5scikit-learn==1.3.2
6joblib==1.3.21# Build image
2docker build -t churn-model:v1.0 .
3
4# Run container
5docker run -d \
6 --name churn-api \
7 -p 8000:8000 \
8 -e MODEL_PATH=/app/models/churn_model.pkl \
9 -e LOG_LEVEL=info \
10 churn-model:v1.0
11
12# Test
13curl http://localhost:8000/health
14
15# View logs
16docker logs churn-api
17
18# Stop container
19docker stop churn-api1# docker-compose.yml
2
3version: '3.8'
4
5services:
6 # Model serving API
7 model-api:
8 build: .
9 ports:
10 - "8000:8000"
11 environment:
12 - MODEL_PATH=/app/models/churn_model.pkl
13 - DATABASE_URL=postgresql://user:pass@db:5432/mlapp
14 - REDIS_URL=redis://redis:6379/0
15 depends_on:
16 - db
17 - redis
18 restart: unless-stopped
19 deploy:
20 resources:
21 limits:
22 cpus: '2'
23 memory: 2G
24 reservations:
25 cpus: '1'
26 memory: 1G
27
28 # PostgreSQL database
29 db:
30 image: postgres:15
31 environment:
32 POSTGRES_USER: user
33 POSTGRES_PASSWORD: pass
34 POSTGRES_DB: mlapp
35 volumes:
36 - postgres_data:/var/lib/postgresql/data
37 ports:
38 - "5432:5432"
39
40 # Redis cache
41 redis:
42 image: redis:7-alpine
43 ports:
44 - "6379:6379"
45 volumes:
46 - redis_data:/data
47
48 # MLflow tracking server
49 mlflow:
50 image: ghcr.io/mlflow/mlflow:v2.8.0
51 ports:
52 - "5000:5000"
53 environment:
54 - BACKEND_STORE_URI=postgresql://user:pass@db:5432/mlflow
55 - DEFAULT_ARTIFACT_ROOT=s3://mlflow-artifacts
56 depends_on:
57 - db
58 command: >
59 mlflow server
60 --backend-store-uri postgresql://user:pass@db:5432/mlflow
61 --default-artifact-root s3://mlflow-artifacts
62 --host 0.0.0.0
63
64volumes:
65 postgres_data:
66 redis_data:Start entire stack:
1docker-compose up -d
2
3# Services available:
4# - Model API: http://localhost:8000
5# - MLflow UI: http://localhost:5000
6# - PostgreSQL: localhost:5432
7# - Redis: localhost:63791# .dockerignore
2__pycache__
3*.pyc
4.git
5.pytest_cache
6notebooks/
7tests/
8*.ipynb
9.envFor GPU inference:
1FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04
2
3# Install Python
4RUN apt-get update && apt-get install -y python3.10 python3-pip
5
6# Install PyTorch with CUDA
7RUN pip install torch==2.1.0+cu118 -f https://download.pytorch.org/whl/torch_stable.html
8
9# ... rest of Dockerfile1# Run with GPU
2docker run --gpus all -p 8000:8000 model:gpuKubernetes (K8s) orchestrates containers at scale: auto-scaling, load balancing, self-healing.
Challenges at scale:
Kubernetes solves these!
Pod: Smallest unit (one or more containers)
Deployment: Manages replicas of Pods
Service: Exposes Pods via stable IP/DNS
ConfigMap/Secret: Configuration management
HPA: Horizontal Pod Autoscaler (scale based on load)
1# deployment.yaml
2
3apiVersion: apps/v1
4kind: Deployment
5metadata:
6 name: churn-model-deployment
7 labels:
8 app: churn-model
9 version: v1
10spec:
11 replicas: 3 # Run 3 instances
12 selector:
13 matchLabels:
14 app: churn-model
15 template:
16 metadata:
17 labels:
18 app: churn-model
19 version: v1
20 spec:
21 containers:
22 - name: model-api
23 image: churn-model:v1.0
24 ports:
25 - containerPort: 8000
26 name: http
27
28 # Resource limits
29 resources:
30 requests:
31 cpu: "500m" # 0.5 CPU
32 memory: "1Gi"
33 limits:
34 cpu: "2" # Max 2 CPUs
35 memory: "4Gi"
36
37 # Health checks
38 livenessProbe: # Restart if fails
39 httpGet:
40 path: /health
41 port: 8000
42 initialDelaySeconds: 30
43 periodSeconds: 10
44
45 readinessProbe: # Remove from load balancer if fails
46 httpGet:
47 path: /health
48 port: 8000
49 initialDelaySeconds: 10
50 periodSeconds: 5
51
52 # Environment from ConfigMap
53 envFrom:
54 - configMapRef:
55 name: model-config
56
57 # Secrets
58 env:
59 - name: DATABASE_PASSWORD
60 valueFrom:
61 secretKeyRef:
62 name: db-secret
63 key: password
64
65 # Volume mounts
66 volumeMounts:
67 - name: model-storage
68 mountPath: /app/models
69
70 volumes:
71 - name: model-storage
72 persistentVolumeClaim:
73 claimName: model-pvc1# service.yaml
2
3apiVersion: v1
4kind: Service
5metadata:
6 name: churn-model-service
7spec:
8 type: LoadBalancer # External access
9 selector:
10 app: churn-model
11 ports:
12 - protocol: TCP
13 port: 80 # External port
14 targetPort: 8000 # Container port
15 sessionAffinity: ClientIP # Sticky sessions1# hpa.yaml
2
3apiVersion: autoscaling/v2
4kind: HorizontalPodAutoscaler
5metadata:
6 name: churn-model-hpa
7spec:
8 scaleTargetRef:
9 apiVersion: apps/v1
10 kind: Deployment
11 name: churn-model-deployment
12
13 minReplicas: 3
14 maxReplicas: 20
15
16 metrics:
17 - type: Resource
18 resource:
19 name: cpu
20 target:
21 type: Utilization
22 averageUtilization: 70 # Scale when CPU > 70%
23
24 - type: Resource
25 resource:
26 name: memory
27 target:
28 type: Utilization
29 averageUtilization: 80
30
31 behavior:
32 scaleDown:
33 stabilizationWindowSeconds: 300 # Wait 5 min before scaling down
34 policies:
35 - type: Percent
36 value: 50 # Scale down max 50% at a time
37 periodSeconds: 601# configmap.yaml
2
3apiVersion: v1
4kind: ConfigMap
5metadata:
6 name: model-config
7data:
8 MODEL_NAME: "churn_prediction_v1"
9 LOG_LEVEL: "info"
10 MAX_BATCH_SIZE: "32"
11 TIMEOUT_SECONDS: "30"
12
13---
14# secret.yaml (base64 encoded values)
15
16apiVersion: v1
17kind: Secret
18metadata:
19 name: db-secret
20type: Opaque
21data:
22 password: cGFzc3dvcmQxMjM= # base64("password123")
23 api_key: YWJjMTIzNDU2Nzg5MA==1# Create namespace
2kubectl create namespace ml-models
3
4# Apply manifests
5kubectl apply -f configmap.yaml -n ml-models
6kubectl apply -f secret.yaml -n ml-models
7kubectl apply -f deployment.yaml -n ml-models
8kubectl apply -f service.yaml -n ml-models
9kubectl apply -f hpa.yaml -n ml-models
10
11# Check status
12kubectl get pods -n ml-models
13kubectl get services -n ml-models
14kubectl get hpa -n ml-models
15
16# View logs
17kubectl logs -f deployment/churn-model-deployment -n ml-models
18
19# Port forward for testing
20kubectl port-forward service/churn-model-service 8000:80 -n ml-models
21
22# Test
23curl http://localhost:8000/healthGradually roll out new version:
1# deployment-v2.yaml (new version)
2
3apiVersion: apps/v1
4kind: Deployment
5metadata:
6 name: churn-model-v2
7spec:
8 replicas: 1 # Start with 1 replica
9 template:
10 spec:
11 containers:
12 - name: model-api
13 image: churn-model:v2.0 # New version
14
15---
16# service-canary.yaml
17
18apiVersion: v1
19kind: Service
20metadata:
21 name: churn-model-service
22spec:
23 selector:
24 app: churn-model # Routes to BOTH v1 and v2
25 ports:
26 - port: 80
27 targetPort: 8000
28
29# Traffic split:
30# v1: 3 replicas = 75% traffic
31# v2: 1 replica = 25% traffic
32
33# Monitor v2 metrics
34# If good: increase v2 replicas, decrease v1
35# If bad: delete v2 deployment (instant rollback)1# gpu-deployment.yaml
2
3apiVersion: apps/v1
4kind: Deployment
5metadata:
6 name: gpu-model
7spec:
8 template:
9 spec:
10 containers:
11 - name: model-api
12 image: model:gpu
13 resources:
14 limits:
15 nvidia.com/gpu: 1 # Request 1 GPU
16
17 nodeSelector:
18 accelerator: nvidia-tesla-v100 # Schedule on V100 nodes1# servicemonitor.yaml
2
3apiVersion: monitoring.coreos.com/v1
4kind: ServiceMonitor
5metadata:
6 name: churn-model-metrics
7spec:
8 selector:
9 matchLabels:
10 app: churn-model
11 endpoints:
12 - port: http
13 path: /metrics
14 interval: 30s1# Add to model_server.py
2from prometheus_client import Counter, Histogram, generate_latest
3
4# Metrics
5REQUEST_COUNT = Counter('model_predictions_total', 'Total predictions')
6REQUEST_LATENCY = Histogram('model_prediction_latency_seconds', 'Prediction latency')
7
8@app.get("/metrics")
9async def metrics():
10 return Response(generate_latest(), media_type="text/plain")
11
12@app.post("/predict")
13@REQUEST_LATENCY.time()
14async def predict(request: PredictionRequest):
15 REQUEST_COUNT.inc()
16 # ... prediction logicMonitoring ensures models perform well in production and detects issues early.
1. Model Performance Metrics
1# Track over time
2- Accuracy / F1 / AUC (if ground truth available)
3- Prediction distribution
4- Confidence scores
5- Class balance
6
7# Alert if:
8- Accuracy drops >5%
9- Prediction distribution changes significantly
10- Too many low-confidence predictions2. System Metrics
1- Request latency (p50, p95, p99)
2- Throughput (requests/second)
3- Error rate (4xx, 5xx)
4- CPU and memory usage
5- GPU utilization (if applicable)
6
7# Alert if:
8- p99 latency > 500ms
9- Error rate > 1%
10- Memory usage > 90%3. Data Drift
1# Input distribution shift
2from scipy.stats import ks_2samp
3
4def detect_drift(training_data, production_data, feature):
5 """
6 Kolmogorov-Smirnov test for distribution shift
7 """
8 statistic, p_value = ks_2samp(
9 training_data[feature],
10 production_data[feature]
11 )
12
13 # p_value < 0.05 → distributions are different
14 if p_value < 0.05:
15 print(f"⚠️ Drift detected in {feature}!")
16 print(f" KS statistic: {statistic:.3f}")
17 return True
18 return False
19
20# Check all features
21for feature in feature_names:
22 detect_drift(X_train, X_production, feature)4. Concept Drift
1# Target distribution shift
2# Example: Customer churn rate increases from 20% to 35%
3
4def detect_concept_drift(y_train, y_recent):
5 """Detect shift in target distribution"""
6 train_mean = np.mean(y_train)
7 recent_mean = np.mean(y_recent)
8
9 # Significant change
10 if abs(recent_mean - train_mean) > 0.1: # >10% change
11 print(f"⚠️ Concept drift detected!")
12 print(f" Training: {train_mean:.2%}")
13 print(f" Recent: {recent_mean:.2%}")
14 return True
15 return False1# monitoring.py
2
3import numpy as np
4from datetime import datetime, timedelta
5from collections import deque
6import logging
7
8logger = logging.getLogger(__name__)
9
10class ModelMonitor:
11 """
12 Production model monitoring system
13
14 Tracks:
15 - Prediction statistics
16 - Data drift
17 - Performance metrics
18 - Alerts
19 """
20
21 def __init__(self, model_name, window_size=1000):
22 self.model_name = model_name
23 self.window_size = window_size
24
25 # Buffers for recent predictions
26 self.predictions = deque(maxlen=window_size)
27 self.confidences = deque(maxlen=window_size)
28 self.latencies = deque(maxlen=window_size)
29 self.timestamps = deque(maxlen=window_size)
30 self.features = deque(maxlen=window_size)
31
32 # Reference statistics (from training)
33 self.reference_stats = None
34
35 # Counters
36 self.total_requests = 0
37 self.total_errors = 0
38
39 def log_prediction(self, features, prediction, confidence,
40 latency, timestamp=None):
41 """Log a prediction for monitoring"""
42 if timestamp is None:
43 timestamp = datetime.now()
44
45 self.predictions.append(prediction)
46 self.confidences.append(confidence)
47 self.latencies.append(latency)
48 self.timestamps.append(timestamp)
49 self.features.append(features)
50
51 self.total_requests += 1
52
53 def log_error(self):
54 """Log an error"""
55 self.total_errors += 1
56
57 def get_stats(self):
58 """Get current statistics"""
59 if len(self.predictions) == 0:
60 return None
61
62 return {
63 'total_requests': self.total_requests,
64 'error_rate': self.total_errors / self.total_requests,
65 'prediction_mean': np.mean(self.predictions),
66 'prediction_std': np.std(self.predictions),
67 'confidence_mean': np.mean(self.confidences),
68 'confidence_min': np.min(self.confidences),
69 'latency_p50': np.percentile(self.latencies, 50),
70 'latency_p95': np.percentile(self.latencies, 95),
71 'latency_p99': np.percentile(self.latencies, 99),
72 'low_confidence_rate': np.mean(np.array(self.confidences) < 0.7)
73 }
74
75 def detect_drift(self, threshold=0.1):
76 """
77 Detect data drift using KS test
78
79 Returns dict of features with drift
80 """
81 if self.reference_stats is None:
82 logger.warning("No reference statistics set")
83 return {}
84
85 if len(self.features) < 100:
86 return {}
87
88 from scipy.stats import ks_2samp
89
90 drift_features = {}
91 recent_features = np.array(list(self.features))
92
93 for i, feature_name in enumerate(self.reference_stats['feature_names']):
94 ref_values = self.reference_stats['features'][:, i]
95 recent_values = recent_features[:, i]
96
97 statistic, p_value = ks_2samp(ref_values, recent_values)
98
99 if p_value < 0.05: # Significant drift
100 drift_features[feature_name] = {
101 'ks_statistic': statistic,
102 'p_value': p_value,
103 'ref_mean': np.mean(ref_values),
104 'recent_mean': np.mean(recent_values),
105 'drift_magnitude': abs(np.mean(recent_values) - np.mean(ref_values))
106 }
107
108 return drift_features
109
110 def check_alerts(self):
111 """
112 Check if any alerts should be triggered
113
114 Returns list of alerts
115 """
116 alerts = []
117 stats = self.get_stats()
118
119 if stats is None:
120 return alerts
121
122 # Error rate alert
123 if stats['error_rate'] > 0.01: # >1% errors
124 alerts.append({
125 'severity': 'high',
126 'type': 'error_rate',
127 'message': f"Error rate {stats['error_rate']:.2%} exceeds threshold",
128 'value': stats['error_rate']
129 })
130
131 # Latency alert
132 if stats['latency_p99'] > 1.0: # >1 second
133 alerts.append({
134 'severity': 'medium',
135 'type': 'latency',
136 'message': f"P99 latency {stats['latency_p99']:.3f}s exceeds threshold",
137 'value': stats['latency_p99']
138 })
139
140 # Low confidence alert
141 if stats['low_confidence_rate'] > 0.2: # >20% low confidence
142 alerts.append({
143 'severity': 'medium',
144 'type': 'confidence',
145 'message': f"{stats['low_confidence_rate']:.1%} predictions have low confidence",
146 'value': stats['low_confidence_rate']
147 })
148
149 # Data drift alert
150 drift_features = self.detect_drift()
151 if len(drift_features) > 0:
152 alerts.append({
153 'severity': 'high',
154 'type': 'data_drift',
155 'message': f"Data drift detected in {len(drift_features)} features",
156 'features': list(drift_features.keys())
157 })
158
159 return alerts
160
161# Usage
162monitor = ModelMonitor("churn_model_v1")
163
164# Set reference statistics (from training data)
165monitor.reference_stats = {
166 'feature_names': ['tenure', 'monthly_charges', ...],
167 'features': X_train_array
168}
169
170# In prediction endpoint
171@app.post("/predict")
172async def predict(request: PredictionRequest):
173 start_time = time.time()
174
175 try:
176 # Make prediction
177 features = extract_features(request)
178 prediction = model.predict(features)
179 confidence = model.predict_proba(features).max()
180
181 # Log for monitoring
182 latency = time.time() - start_time
183 monitor.log_prediction(
184 features, prediction, confidence,
185 latency, datetime.now()
186 )
187
188 return {"prediction": prediction, "confidence": confidence}
189
190 except Exception as e:
191 monitor.log_error()
192 raise
193
194# Periodic check (every 5 minutes)
195@app.on_event("startup")
196async def start_monitoring():
197 async def check_health():
198 while True:
199 await asyncio.sleep(300) # 5 minutes
200
201 # Get stats
202 stats = monitor.get_stats()
203 logger.info(f"Model stats: {stats}")
204
205 # Check alerts
206 alerts = monitor.check_alerts()
207 for alert in alerts:
208 logger.warning(f"🚨 ALERT: {alert}")
209
210 # Send to alerting system (PagerDuty, Slack)
211 send_alert(alert)
212
213 asyncio.create_task(check_health())1# alerts.py
2
3import requests
4
5def send_slack_alert(alert):
6 """Send alert to Slack"""
7 webhook_url = os.getenv("SLACK_WEBHOOK_URL")
8
9 severity_emoji = {
10 'high': '🚨',
11 'medium': '⚠️',
12 'low': 'ℹ️'
13 }
14
15 message = {
16 "text": f"{severity_emoji[alert['severity']]} {alert['message']}",
17 "attachments": [{
18 "color": "danger" if alert['severity'] == 'high' else "warning",
19 "fields": [{
20 "title": "Model",
21 "value": "churn_model_v1",
22 "short": True
23 }, {
24 "title": "Type",
25 "value": alert['type'],
26 "short": True
27 }]
28 }]
29 }
30
31 requests.post(webhook_url, json=message)
32
33def send_pagerduty_alert(alert):
34 """Send high-severity alerts to PagerDuty"""
35 if alert['severity'] != 'high':
36 return
37
38 api_key = os.getenv("PAGERDUTY_API_KEY")
39 # ... PagerDuty integrationPrometheus queries:
1# Request rate
2rate(model_predictions_total[5m])
3
4# Error rate
5rate(http_requests_total{status=~"5.."}[5m]) /
6rate(http_requests_total[5m])
7
8# P99 latency
9histogram_quantile(0.99, rate(model_prediction_latency_seconds_bucket[5m]))
10
11# Low confidence rate
12rate(predictions_low_confidence_total[5m]) /
13rate(model_predictions_total[5m])1# ab_test.py
2
3import random
4
5class ABTest:
6 """
7 A/B test framework for model comparison
8 """
9
10 def __init__(self, model_a, model_b, traffic_split=0.5):
11 self.model_a = model_a
12 self.model_b = model_b
13 self.traffic_split = traffic_split
14
15 # Track metrics
16 self.metrics = {
17 'a': {'requests': 0, 'errors': 0, 'latencies': []},
18 'b': {'requests': 0, 'errors': 0, 'latencies': []}
19 }
20
21 def predict(self, features):
22 """Route request to A or B"""
23 # Assign to variant
24 variant = 'a' if random.random() < self.traffic_split else 'b'
25 model = self.model_a if variant == 'a' else self.model_b
26
27 # Make prediction
28 start_time = time.time()
29 try:
30 prediction = model.predict(features)
31 latency = time.time() - start_time
32
33 self.metrics[variant]['requests'] += 1
34 self.metrics[variant]['latencies'].append(latency)
35
36 return prediction, variant
37
38 except Exception as e:
39 self.metrics[variant]['errors'] += 1
40 raise
41
42 def get_results(self):
43 """Compare variants"""
44 return {
45 'variant_a': {
46 'requests': self.metrics['a']['requests'],
47 'error_rate': self.metrics['a']['errors'] / max(1, self.metrics['a']['requests']),
48 'avg_latency': np.mean(self.metrics['a']['latencies'])
49 },
50 'variant_b': {
51 'requests': self.metrics['b']['requests'],
52 'error_rate': self.metrics['b']['errors'] / max(1, self.metrics['b']['requests']),
53 'avg_latency': np.mean(self.metrics['b']['latencies'])
54 }
55 }Congratulations! You've completed the Python AI & Machine Learning course. You now have end-to-end skills from fundamentals to production deployment! 🎉
Build an end-to-end MLOps pipeline with experiment tracking, model serving API, Docker containerization, and production monitoring. This comprehensive example demonstrates the complete journey from model training to production deployment.
1# Complete MLOps Pipeline Example
2# End-to-end: Training → Tracking → Serving → Monitoring
3
4# See the interactive editor for the complete implementation!
5# This example covers:
6
7# 1. Experiment Tracking with MLflow
8# - Log parameters, metrics, and artifacts
9# - Compare multiple runs
10# - Select best model
11
12# 2. Model Serving API with FastAPI
13# - RESTful endpoints
14# - Request validation
15# - Error handling
16# - Interactive docs
17
18# 3. Production Monitoring
19# - Data drift detection
20# - Performance tracking
21# - Alert system
22# - Logging
23
24# 4. Deployment Artifacts
25# - Dockerfile for containerization
26# - Kubernetes manifests
27# - Docker Compose for local development
28
29print("Navigate to the Editor tab to see the complete implementation!")
30print()
31print("The complete MLOps pipeline includes:")
32print(" ✓ Experiment tracking (MLflow)")
33print(" ✓ Model serving API (FastAPI)")
34print(" ✓ Monitoring & drift detection")
35print(" ✓ Docker containerization")
36print(" ✓ Kubernetes deployment")
37print()
38print("🚀 Ready for production deployment!")This comprehensive MLOps example demonstrates the complete production ML lifecycle.
1. Complete MLOps Workflow
The implementation covers every stage of production ML:
2. Industry Best Practices
Experiment Tracking:
Model Serving:
Production Monitoring:
3. Deployment Patterns
Local Development:
1# Docker Compose for local testing
2docker-compose up -d
3# Services: API, PostgreSQL, Redis, MLflowProduction Deployment:
1# Kubernetes with auto-scaling
2kubectl apply -f deployment.yaml
3# Features: Load balancing, health checks, HPA4. Real-World Challenges Addressed
Reproducibility:
Scalability:
Reliability:
Observability:
5. Production Monitoring
The monitoring system detects:
Data Drift:
Performance Degradation:
System Health:
6. MLOps Maturity Progression
Level 0 → Level 3 (Implemented in this example):
Level 0 (Manual):
Level 1 (Automated CI/CD):
Level 2 (Automated Training):
Level 3 (Automated Deployment):
1. MLOps is Essential
2. Automation is Key
3. Monitor Everything
4. Start Simple, Iterate
5. Tools Don't Matter (Principles Do)
What matters: reproducibility, monitoring, automation, reliability.
You now understand:
Next Steps:
Congratulations! You've completed the Python AI & Machine Learning course. You have the skills to build, deploy, and maintain production AI systems. 🎉🚀
🎉 Congratulations! You've completed the Python AI & Machine Learning course!
You've mastered the complete AI/ML journey:
✅ Foundations (Lessons 1-2): AI concepts, mathematics, linear algebra, calculus, probability
✅ Deep Learning Frameworks (Lessons 3-6): TensorFlow, PyTorch, Neural Networks, Keras
✅ Computer Vision (Lessons 7-8, 12-13): CNNs, ResNet, Inception, Object Detection, Segmentation
✅ Sequential Models (Lessons 9-10): RNNs, LSTMs, GRU, Attention, Seq2Seq
✅ Transformers & NLP (Lessons 11, 14-15): Self-attention, BERT, GPT, Tokenization, Fine-tuning
✅ Generative AI (Lessons 16-17): GANs, VAEs, DCGAN, Latent Space, Image Generation
✅ Reinforcement Learning (Lessons 18-19): Q-Learning, DQN, Policy Gradients, PPO, Actor-Critic
✅ Production Deployment (Lesson 20): MLOps, Docker, Kubernetes, Monitoring, CI/CD
1. Build Real-World Projects
2. Contribute to Open Source
3. Stay Current (October 2025)
4. Join Communities
5. Specialize
ML Engineer: Build production ML systems
Research Scientist: Advance state-of-the-art
MLOps Engineer: Scale ML infrastructure
AI Product Manager: Define ML product strategy
Data Scientist: Extract insights + build models
You have the knowledge to:
The AI revolution is here. You're equipped to be part of it. 🚀
Thank you for completing this course. Now go build something amazing! 💪
Your progress is saved in this browser, on this device only.