Practice Project 21

MLOps Deployment Pipeline Project

Deploy trained models through containerized FastAPI endpoints. Implement statistical validation checks to monitor incoming data stream distributions for drift.

Domain / Environment

MLOps / Docker / Conda VM

Difficulty

Advanced (4/5)

Course Module

MLOps & Deployment

Deliverables

FastAPI app file, Dockerfile configuration, & drift logs

1. Containerized Model-Serving & Monitoring Architecture

The diagram below displays the containerized MLOps architecture. The serialized model is deployed inside a FastAPI Docker container, exposing prediction endpoints. Incoming request inputs are monitored by a drift detection script to compare distributions against training data baseline metrics.

2. Part 1: Step-by-Step Action Items & Key Execution Steps

STEP 1

Activate Python Virtual Environment

Point the terminal execution environment to the course conda sandbox environment.

$ conda activate ds_ai_ml

This targets active python libraries to the isolated virtual sandbox.

STEP 2

Create Project Folders inside Linux VM

Create a dedicated folder for the project files inside your guest VM home folder directory.

$ mkdir -p ~/Projects/mlops_pipeline && cd ~/Projects/mlops_pipeline

This sets up the working directory layout for the MLOps code files.

STEP 3

Install FastAPI and Uvicorn via Pip

Install the required API serving packages inside the active conda session.

$ pip install fastapi uvicorn scipy scikit-learn pandas numpy joblib pydantic

This installs the FastAPI application framework, Pydantic data schemas, SciPy statistical tools, and Uvicorn servers.

STEP 4

Create FastAPI backend file in VS Code

Launch VS Code and create the FastAPI prediction script file.

Launch VS Code via terminal "code ." -> New File -> Type: main.py -> Paste Python code -> Save file

This registers the FastAPI web application, endpoint routes, and prediction model load actions in `main.py`.

STEP 5

Create Docker configuration file in VS Code

Create a Dockerfile configuration script to containerize the FastAPI web application.

New File -> Type: Dockerfile -> Paste configuration script -> Save file

This registers base image configurations, copies source code files, and exposes application network ports.

STEP 6

Launch and test API endpoints

Start the Uvicorn local server instance to verify FastAPI endpoints.

$ uvicorn main:app --reload

This starts the development server, exposing endpoints locally on port 8000.

STEP 7

Test API Endpoints using Curl

Open a secondary terminal window and submit mock request payloads to verify predictions.

$ curl -X POST "http://127.0.0.1:8000/predict" -H "Content-Type: application/json" -d '{"age": 28, "salary": 72000.0, "city": "London"}'

This returns the prediction response JSON containing prediction label classes.

STEP 8

Run the Data Drift Monitor

Execute the drift detection script to identify distribution shifts in production features.

$ python monitor_drift.py

This runs statistical Kolmogorov-Smirnov tests and prints drift flags to the console.

3. MLOps Execution Flow

The flowchart below outlines the MLOps pipeline execution flow. It details the steps from training and serializing the model to exposing endpoints via FastAPI and containerizing the app with Docker.

4. Part 2: Complete Deliverable Assets & Production Templates

To run the MLOps serving and monitoring setup, we need the FastAPI script, Dockerfile config, and drift detection script. Below is a line-by-line explanation of the code, followed by the combined template files.

Step-by-Step Code Construction

Lines 1 - 8

Import API and schema modules

Include system packages, FastAPI endpoints, Pydantic schemas, and model loading utilities in the script.

from fastapi import FastAPI from pydantic import BaseModel import joblib import pandas as pd import numpy as np

These imports pull core FastAPI classes, Pydantic model configurations, and joblib model deserializers.

Lines 9 - 18

Define Input Data Schema

Define a Pydantic model class to validate the structured request JSON payloads submitted by clients.

class PredictionInput(BaseModel): age: float salary: float city: str

This sets up schema verification rules to enforce correct input types for age, salary, and city features.

Lines 19 - 30

FastAPI Endpoints

Expose `/predict` POST endpoints and `/health` GET check endpoints in the FastAPI app class instance.

app = FastAPI() @app.post("/predict") def predict(payload: PredictionInput): # Preprocess and score payload values return {"prediction": 0}

This sets up endpoint routing to accept inputs, process columns, and return prediction responses.

Lines 31 - 45

Implement Kolmogorov-Smirnov Drift Check

Calculate numerical feature distributions and run statistical Kolmogorov-Smirnov tests to detect data drift.

from scipy.stats import ks_2samp statistic, p_val = ks_2samp(baseline_data, production_data) if p_val < 0.05: print("ALERT: Feature distribution drift detected!")

This compares production input distributions against baseline training metrics, flagging drift alerts when `p_value` drops below the 0.05 threshold.

Production templates

1. FastAPI serving code (Save as ~/Projects/mlops_pipeline/main.py):

# main.py - FastAPI endpoint serving predictive models from fastapi import FastAPI from pydantic import BaseModel import joblib import pandas as pd import numpy as np import os app = FastAPI(title="MLOps serving model API") class PredictionInput(BaseModel): age: float salary: float city: str # Load pre-trained preprocessor pipeline locally (simulated dummy if file missing) preprocessor_path = "../python_preproc/preprocessor.joblib" if os.path.exists(preprocessor_path): preprocessor = joblib.load(preprocessor_path) else: preprocessor = None @app.get("/health") def health_check(): return {"status": "healthy", "preprocessor_loaded": preprocessor is not None} @app.post("/predict") def predict(payload: PredictionInput): print(f"Incoming request details: {payload}") # Log request features to a file for drift monitoring with open("requests.log", "a") as log_file: log_file.write(f"{payload.age},{payload.salary},{payload.city}\n") # Simulated prediction inference # (If using real classifier model weights, run preprocessor.transform() first) prob = 1.0 / (1.0 + np.exp(-(-3.0 + 0.05 * payload.age + 0.00001 * payload.salary))) pred_label = 1 if prob > 0.5 else 0 return { "churn_probability": float(prob), "prediction_label": int(pred_label) } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)

2. Docker Configuration script (Save as ~/Projects/mlops_pipeline/Dockerfile):

FROM python:3.9-slim WORKDIR /app # Install system dependencies RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/* # Install python libraries RUN pip install --no-cache-dir fastapi uvicorn scipy pandas numpy joblib pydantic COPY . /app EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

3. Python Drift Detection script (Save as ~/Projects/mlops_pipeline/monitor_drift.py):

# monitor_drift.py - Check baseline vs production inputs for distribution drift import numpy as np from scipy.stats import ks_2samp def check_drift(baseline, production, feature_name): # Run Kolmogorov-Smirnov test to compare distributions stat, p_val = ks_2samp(baseline, production) print(f"Feature: {feature_name:10} | KS statistic: {stat:.4f} | p-value: {p_val:.4f}") if p_val < 0.05: print(f"ALERT: Distribution drift detected for feature '{feature_name}'! (p-value < 0.05)") else: print(f"PASSED: No drift detected for feature '{feature_name}'.") def main(): print("=== Starting MLOps Data Drift Monitoring Check ===") np.random.seed(42) # Training dataset baseline (e.g. mean age = 35) baseline_age = np.random.normal(loc=35, scale=8, size=200) # Scenario A: Stable production traffic (mean age = 35.5) stable_prod_age = np.random.normal(loc=35.5, scale=8, size=100) # Scenario B: Driffed production traffic (mean age = 45 - older demographic) drifted_prod_age = np.random.normal(loc=45, scale=8, size=100) print("\n--- Testing Scenario A: Stable Production Traffic ---") check_drift(baseline_age, stable_prod_age, "Age") print("\n--- Testing Scenario B: Shifted Production Traffic ---") check_drift(baseline_age, drifted_prod_age, "Age") print("\n=== MLOps Data Drift Monitoring Completed! ===") if __name__ == "__main__": main()

5. Deliverables Summary

Verify that the following configurations and outputs exist inside your project workspace.

Created Files / Templates

~/Projects/mlops_pipeline/main.py - FastAPI backend api service file.
~/Projects/mlops_pipeline/Dockerfile - Docker container configuration file.
~/Projects/mlops_pipeline/monitor_drift.py - Statistical drift monitor script.

Verification Artifacts / Execution Proof

FastAPI web app starting and running successfully on port 8000.
Inference POST requests returning prediction response JSON records.
Data drift check scripts logging KS test stats and alert flags.

6. Closing Explanation: Why We Did This & What It Accomplishes

Architectural Intent & Operational Impact

Why We Did This

FastAPI exposes predictive models as APIs, enabling client applications to query them.
Docker containers package application runtimes to ensure consistent behavior across staging and production.
Drift detection scripts flag distribution shifts in incoming feature streams, preventing model degradation.

What This Accomplishes

Exposes model prediction logic via FastAPI endpoints.
Exposes docker configuration scripts.
Implements statistical Kolmogorov-Smirnov test monitoring checks.