Practice Questions — MLOps and Model Deployment

pipeline_stages = ["Data", "Preprocess", "Train", "Evaluate", "Deploy", "Monitor"]
for i, stage in enumerate(pipeline_stages, 1):
    print(f"{i}. {stage}")

save_methods = {
    "pickle": ".pkl",
    "joblib": ".joblib",
    "keras": ".keras",
    "torch": ".pt"
}
for method, ext in save_methods.items():
    print(f"{method}: saves as *{ext}")

serving_patterns = [
    ("Real-time", "< 100ms", "Chatbots, search"),
    ("Batch", "Minutes-hours", "Reports, emails"),
    ("Streaming", "Seconds", "Fraud detection")
]
for pattern, latency, use_case in serving_patterns:
    print(f"{pattern:12s} | {latency:15s} | {use_case}")

import json

# Simulated API request and response
request_data = {"features": [0.5, -1.2, 3.4, 0.8]}
response_data = {"prediction": 1, "confidence": 0.92}

print(f"Request: {json.dumps(request_data)}")
print(f"Response: {json.dumps(response_data)}")
print(f"Feature count: {len(request_data['features'])}")
print(f"Prediction: {'positive' if response_data['prediction'] == 1 else 'negative'}")

docker_layers = [
    ("FROM python:3.11-slim", "Base image"),
    ("COPY requirements.txt .", "Dependencies spec"),
    ("RUN pip install -r requirements.txt", "Install deps (cached)"),
    ("COPY app.py .", "Application code"),
    ("COPY model.joblib .", "ML model"),
    ("CMD uvicorn app:app", "Start command")
]

print("Docker Layer Order (optimized for caching):")
for i, (cmd, desc) in enumerate(docker_layers, 1):
    print(f"  {i}. {desc:25s} -> {cmd}")

print(f"\nTotal layers: {len(docker_layers)}")
print("Note: Change to app.py rebuilds only layers 4-6 (fast rebuild)")

import numpy as np
from scipy import stats

np.random.seed(42)

# Reference distribution (training)
ref = np.random.normal(0, 1, 1000)

# Test 1: Same distribution (no drift)
test_same = np.random.normal(0, 1, 1000)
stat1, p1 = stats.ks_2samp(ref, test_same)

# Test 2: Different distribution (drift!)
test_diff = np.random.normal(0.5, 1.5, 1000)
stat2, p2 = stats.ks_2samp(ref, test_diff)

print(f"Same distribution:  KS={stat1:.4f}, p={p1:.4f}, drift={'YES' if p1 < 0.05 else 'No'}")
print(f"Drifted distribution: KS={stat2:.4f}, p={p2:.4f}, drift={'YES' if p2 < 0.05 else 'No'}")

class ModelVersionManager:
    def __init__(self):
        self.versions = []
        self.active = None
    
    def register(self, name, accuracy):
        version = {"name": name, "accuracy": accuracy, "version": len(self.versions) + 1}
        self.versions.append(version)
        return version["version"]
    
    def promote(self, version_num):
        for v in self.versions:
            if v["version"] == version_num:
                self.active = v
                return True
        return False
    
    def get_active(self):
        return self.active

mgr = ModelVersionManager()
mgr.register("rf_v1", 0.89)
mgr.register("rf_v2", 0.92)
v3 = mgr.register("gb_v1", 0.95)

mgr.promote(v3)
active = mgr.get_active()
print(f"Total versions: {len(mgr.versions)}")
print(f"Active: {active['name']} (v{active['version']}, acc={active['accuracy']})")

checks = ["Data drift", "Concept drift", "Accuracy decay", "Latency spike", "Error rate"]
for c in checks:
    print(f"  Monitor: {c}")
print(f"Total: {len(checks)}")

import json

def validate_request(data, n_features=4):
    if data is None: return "Invalid JSON"
    if "features" not in data: return "Missing features"
    if len(data["features"]) != n_features: return f"Expected {n_features}, got {len(data['features'])}"
    return "OK"

requests = [{"features":[1,2,3,4]}, {"data":[1]}, {"features":[1,2]}, None]
for req in requests:
    print(f"{str(req)[:35]:35s} -> {validate_request(req)}")

def api_capacity(latency_ms, workers):
    return workers * (1000 / latency_ms)

for latency, workers in [(10, 4), (50, 4), (200, 4), (50, 16)]:
    rps = api_capacity(latency, workers)
    print(f"{latency:3d}ms, {workers:2d} workers -> {rps:.0f} RPS")

mlops_tools = {"Versioning": "MLflow", "Container": "Docker", "API": "FastAPI", "CI/CD": "GitHub Actions"}
for purpose, tool in mlops_tools.items():
    print(f"{purpose:12s} -> {tool}")

cloud_options = [
    ("AWS SageMaker", "Amazon"),
    ("Vertex AI", "Google"),
    ("Azure ML", "Microsoft"),
    ("HF Spaces", "Hugging Face")
]
for service, provider in cloud_options:
    print(f"{provider:15s} -> {service}")

http_codes = {
    200: "OK -- prediction returned",
    400: "Bad Request -- invalid input",
    404: "Not Found -- wrong endpoint",
    500: "Server Error -- model failed"
}
for code, meaning in http_codes.items():
    print(f"HTTP {code}: {meaning}")

import os

# Simulated file sizes for model artifacts
artifacts = {
    "model.pkl": 5242880,       # 5 MB
    "vectorizer.pkl": 1048576,  # 1 MB
    "config.json": 1024,        # 1 KB
    "requirements.txt": 256     # 256 bytes
}

total = 0
for name, size in artifacts.items():
    mb = size / (1024 * 1024)
    print(f"{name:20s}: {mb:8.2f} MB")
    total += size

print(f"{'Total':20s}: {total/(1024*1024):8.2f} MB")

def api_latency_report(latencies_ms):
    import numpy as np
    p50 = np.percentile(latencies_ms, 50)
    p95 = np.percentile(latencies_ms, 95)
    p99 = np.percentile(latencies_ms, 99)
    avg = np.mean(latencies_ms)
    return {"avg": round(avg, 1), "p50": round(p50, 1), "p95": round(p95, 1), "p99": round(p99, 1)}

import numpy as np
np.random.seed(42)
latencies = np.random.exponential(scale=15, size=1000)  # Exponential distribution
report = api_latency_report(latencies)

for metric, value in report.items():
    sla_ok = value < 100
    print(f"{metric:4s}: {value:6.1f}ms {'OK' if sla_ok else 'BREACH'}")

import mlflow

class MockMLflowRun:
    def __init__(self, params, metrics, model_name):
        self.params = params
        self.metrics = metrics
        self.model_name = model_name

runs = [
    MockMLflowRun({"n_estimators": "100", "max_depth": "5"}, {"accuracy": 0.89, "f1": 0.87}, "rf_v1"),
    MockMLflowRun({"n_estimators": "200", "max_depth": "10"}, {"accuracy": 0.93, "f1": 0.91}, "rf_v2"),
    MockMLflowRun({"n_estimators": "50", "learning_rate": "0.1"}, {"accuracy": 0.95, "f1": 0.94}, "gb_v1")
]

print("MLflow Experiment Comparison:")
print(f"{'Model':8s} | {'Accuracy':10s} | {'F1':10s} | {'Params':30s}")
print("-" * 65)
for run in runs:
    params_str = ", ".join(f"{k}={v}" for k, v in run.params.items())
    print(f"{run.model_name:8s} | {run.metrics['accuracy']:10.4f} | {run.metrics['f1']:10.4f} | {params_str}")

best = max(runs, key=lambda r: r.metrics["accuracy"])
print(f"\nBest model: {best.model_name} (accuracy={best.metrics['accuracy']})")

deployment_checklist = [
    "Model serialized (joblib/pickle)",
    "API endpoint created (/predict)",
    "Input validation added",
    "Health check endpoint (/health)",
    "Dependencies pinned (requirements.txt)",
    "Dockerized",
    "Monitoring configured"
]
print(f"Deployment Checklist ({len(deployment_checklist)} items):")
for i, item in enumerate(deployment_checklist, 1):
    print(f"  [{i}] {item}")

def compare_frameworks(name, features):
    return f"{name}: {', '.join(features)}"

flask = compare_frameworks("Flask", ["Simple", "Synchronous", "Manual validation"])
fastapi = compare_frameworks("FastAPI", ["Modern", "Async", "Auto validation", "Auto docs"])

print(flask)
print(fastapi)
print(f"\nFlask features: 3")
print(f"FastAPI features: 4")
print(f"Winner for ML APIs: FastAPI")