Naman

Posted on Sep 18 AI-assisted

Building a Production-Grade End-to-End MLOps Pipeline from Scratch

#machinelearning #mlops #python #devops

Building a Production-Grade End-to-End MLOps Pipeline from Scratch

Most machine learning tutorials end at model.fit(). But in the real world, training a model is barely 10% of the work. The remaining 90% is everything that makes ML actually work in production — data versioning, validation, experiment tracking, API serving, automated testing, containerization, CI/CD, monitoring, and drift detection.

In this walkthrough, I'll take you through building a complete, production-grade MLOps pipeline using the classic Titanic survival prediction problem. The focus isn't on building a complex model — it's on implementing every production practice that separates a Jupyter notebook experiment from a deployable ML system.

GitHub Repository: End-to-end-MLOPS-Pipeline

Architecture Overview

Here's the complete system we're building:

Data (Titanic CSV)
 │
 ▼
DVC (Data Versioning)
 │
 ▼
Data Validation (Schema & Constraint Checks)
 │
 ▼
Scikit-learn Training Pipeline
 │
 ▼
MLflow (Experiment Tracking & Model Registry)
 │
 ▼
Pytest (Automated Testing)
 │
 ▼
Docker (Containerization)
 │
 ▼
GitHub Actions (CI/CD)
 │
 ▼
FastAPI (Inference Service)
 │
 ├── Prometheus (Monitoring)
 │
 └── Evidently AI (Drift Detection)

Technology Stack

Component	Tool
Version Control	Git, GitHub
Data Versioning	DVC
Model Training	Scikit-learn
Experiment Tracking	MLflow
API Framework	FastAPI
Testing	Pytest
Containerization	Docker
CI/CD	GitHub Actions
Monitoring	Prometheus
Drift Detection	Evidently AI

Project Structure

End-to-end-MLOPS-Pipeline/
│
├── data/
│   └── raw/
│       └── titanic.csv
│
├── models/
│   └── model.joblib
│
├── src/
│   ├── data_ingestion.py
│   ├── data_validation.py
│   ├── data_preprocess.py
│   ├── train.py
│   └── drift_detection.py
│
├── api/
│   ├── __init__.py
│   └── main.py
│
├── tests/
│   ├── __init__.py
│   ├── test_api.py
│   └── test_model.py
│
├── monitoring/
│   ├── prometheus.yml
│   └── drift_report.html
│
├── .github/
│   └── workflows/
│       └── ci_cd.yml
│
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── .gitignore
├── .dockerignore
└── .dvc/

Step 1: Project Initialization & Environment Setup

Start by initializing Git and creating a virtual environment:

git init
python -m venv venv

# Activate the virtual environment
# Windows PowerShell:
.\venv\Scripts\Activate.ps1
# Linux/macOS:
source venv/bin/activate

`.gitignore`

Create a .gitignore that keeps our repository clean:

# Virtual environments
venv/
env/
.env

# Python cache
__pycache__/
*.pyc

# Data (tracked by DVC, not Git)
data/raw/*
!data/raw/*.dvc
!data/raw/*.csv
data/processed/

# Models and MLflow
models/
mlruns/

`requirements.txt`

# Data & ML
pandas>=2.0.0
numpy>=1.24.0
scikit-learn>=1.3.0
joblib>=1.3.0

# Data Validation & Tracking
dvc>=3.0.0
mlflow>=2.10.0

# API
fastapi>=0.100.0
uvicorn[standard]>=0.22.0
pydantic>=2.0.0

# Testing
pytest>=7.4.0
httpx>=0.24.0
prometheus-client>=0.17.0
evidently

Install everything:

pip install -r requirements.txt

Step 2: Data Versioning with DVC

Why not just commit datasets to Git?

Git was designed for source code — small text files. Datasets can be hundreds of megabytes or gigabytes. Committing them directly to Git makes your repository bloated, slow to clone, and impossible to manage at scale.

DVC (Data Version Control) solves this by:

Storing lightweight .dvc metadata files in Git (a few bytes).
Storing actual data files in remote storage (S3, GCS, local cache).
Allowing you to roll back datasets to any historical version, just like git checkout.

Setup

Download the Titanic dataset and place it in data/raw/titanic.csv.

# Initialize DVC
dvc init

# Track the dataset
dvc add data/raw/titanic.csv

# Commit the metadata
git add .
git commit -m "Initialize project and track dataset with DVC"

DVC creates a data/raw/titanic.csv.dvc file — a tiny metadata pointer that Git tracks instead of the actual CSV.

Step 3: Data Ingestion (`src/data_ingestion.py`)

The data ingestion module is responsible for loading raw data from disk with defensive checks — ensuring the file exists and isn't empty before passing it downstream.

import os
import logging
import pandas as pd

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)

def load_data(file_path: str) -> pd.DataFrame:
    """
    Load raw CSV data and validate it.
    """
    if not os.path.exists(file_path):
        logging.error(f"File not found at path: {file_path}")
        raise FileNotFoundError(f"File does not exist: {file_path}")

    logging.info(f"Loading data from {file_path}")
    df = pd.read_csv(file_path)

    if df.empty:
        logging.error("Loaded dataframe is empty")
        raise ValueError("Dataframe is empty")

    logging.info(f"Successfully loaded dataset with shape: {df.shape}")
    return df


if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    print(df.head())

Why `logging` instead of `print()`?

In production MLOps, print() statements disappear into the void. logging gives you:

Timestamps for every action.
Severity levels (INFO, WARNING, ERROR) to filter noise.
The ability to write logs to files or send them to monitoring systems.

Step 4: Data Validation (`src/data_validation.py`)

Before touching any ML model, we validate that the incoming data matches our expectations. This is the first line of defense against silent model failures caused by upstream data changes.

import os
import sys
import logging
import pandas as pd

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

# Define expected schema and column rules
REQUIRED_COLUMNS = [
    "Survived", "Pclass", "Name", "Sex", "Age", 
    "SibSp", "Parch", "Ticket", "Fare", "Embarked"
]
VALID_PCLASS = {1, 2, 3}
VALID_SEX = {"male", "female"}


def validate_data(df: pd.DataFrame) -> bool:
    """
    Validates dataset integrity, schema, and column value constraints.
    """
    logging.info("Starting data validation checks...")

    # 1. Check for required columns
    missing_cols = [col for col in REQUIRED_COLUMNS if col not in df.columns]
    if missing_cols:
        logging.error(f"Validation failed: Missing columns: {missing_cols}")
        raise ValueError(f"Missing required columns: {missing_cols}")

    # 2. Check categorical domains
    invalid_pclass = set(df["Pclass"].dropna().unique()) - VALID_PCLASS
    if invalid_pclass:
        logging.error(f"Validation failed: Unexpected Pclass values: {invalid_pclass}")
        raise ValueError(f"Invalid values in Pclass: {invalid_pclass}")

    invalid_sex = set(df["Sex"].dropna().unique()) - VALID_SEX
    if invalid_sex:
        logging.error(f"Validation failed: Unexpected Sex values: {invalid_sex}")
        raise ValueError(f"Invalid values in Sex: {invalid_sex}")

    # 3. Check target column for missing values
    if df["Survived"].isnull().any():
        logging.error("Validation failed: Target column 'Survived' contains null values.")
        raise ValueError("Target column 'Survived' cannot contain null values.")

    # 4. Range checks
    if (df["Fare"] < 0).any():
        logging.error("Validation failed: Found negative values in 'Fare'.")
        raise ValueError("Fare values cannot be negative.")

    logging.info("All data validation checks passed successfully!")
    return True


if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    validate_data(df)

What are we defending against?

Check	Why It Matters
Required columns exist	If someone upstream renames `Sex` to `Gender`, your pipeline fails here — not deep inside model training with a cryptic error.
`Pclass` ∈ {1, 2, 3}	Unknown categories would crash one-hot encoding or produce garbage features.
`Survived` has no nulls	Your target variable for supervised learning must be complete.
`Fare >= 0`	Negative fares are logically impossible and indicate data corruption.

Run it:

python src/data_validation.py
# Output: All data validation checks passed successfully!

Step 5: Preprocessing & Feature Engineering (`src/data_preprocess.py`)

import os
import sys
import logging
import pandas as pd
from sklearn.model_selection import train_test_split

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data
from src.data_validation import validate_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def preprocess_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    Cleans raw titanic data using simple pandas operations:
    1. Fills missing values (Age with median, Embarked with mode).
    2. Drops columns that aren't useful (PassengerId, Name, Ticket, Cabin).
    3. Converts Sex and Embarked to numbers using one-hot encoding (get_dummies).
    """
    df = df.copy()

    # 1. Fill missing values
    df["Age"] = df["Age"].fillna(df["Age"].median())
    df["Fare"] = df["Fare"].fillna(df["Fare"].median())
    df["Embarked"] = df["Embarked"].fillna(df["Embarked"].mode()[0])

    # 2. Drop columns not needed for modeling
    drop_cols = ["PassengerId", "Name", "Ticket", "Cabin"]
    df = df.drop(columns=[col for col in drop_cols if col in df.columns])

    # 3. Convert text/categorical columns to numbers (One-Hot Encoding)
    df = pd.get_dummies(df, columns=["Sex", "Embarked"], drop_first=True, dtype=int)

    return df


def split_data(df: pd.DataFrame, test_size: float = 0.2, random_state: int = 42):
    """
    Splits the cleaned dataframe into train and test sets.
    """
    X = df.drop(columns=["Survived"])
    y = df["Survived"]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=test_size, random_state=random_state, stratify=y
    )

    return X_train, X_test, y_train, y_test


if __name__ == "__main__":
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)
    validate_data(df)

    cleaned_df = preprocess_data(df)
    logging.info(f"Cleaned data shape: {cleaned_df.shape}")
    print("\nCleaned Data Preview:")
    print(cleaned_df.head())

    X_train, X_test, y_train, y_test = split_data(cleaned_df)
    logging.info(f"Train samples: {len(X_train)}, Test samples: {len(X_test)}")

Key Decisions Explained

fillna with median/mode: Median is robust to outliers (unlike mean). Mode fills categorical gaps with the most common value.
Dropped columns: PassengerId is a unique identifier (no predictive value). Name and Ticket are high-cardinality text. Cabin is >70% null.
pd.get_dummies(drop_first=True): Converts Sex (male/female) → Sex_male (1/0) and Embarked (C/Q/S) → Embarked_Q, Embarked_S. The drop_first avoids the dummy variable trap.
stratify=y: Ensures both train and test sets maintain the same 38/62% survived/died ratio.

Run it:

python src/data_preprocess.py
# Output:
# Cleaned data shape: (891, 9)
# Train samples: 712, Test samples: 179

Step 6: Model Training & Experiment Tracking with MLflow (`src/train.py`)

This is where most tutorials stop. But we're not just training a model — we're logging every experiment so we can compare runs, reproduce results, and promote the best model to production.

import os
import sys
import logging
import joblib
import mlflow
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import (
    accuracy_score,
    precision_score,
    recall_score,
    f1_score
)

sys.path.append(
    os.path.abspath(
        os.path.join(os.path.dirname(__file__), "..")
    )
)
from src.data_ingestion import load_data
from src.data_preprocess import preprocess_data, split_data

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)


def train_model(
    n_estimators: int = 100,
    max_depth: int = 5,
    random_state: int = 42
):
    """
    Train a Random Forest model and log results to MLflow.
    """
    # Load raw data
    raw_data_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_data_path)

    # Preprocess data
    cleaned_df = preprocess_data(df)

    # Train-test split
    X_train, X_test, y_train, y_test = split_data(
        cleaned_df,
        random_state=random_state
    )

    # Configure MLflow
    mlflow.set_tracking_uri("sqlite:///mlflow.db")
    mlflow.set_experiment("Titanic-Survival-Prediction")

    with mlflow.start_run():
        logging.info("Training Random Forest Classifier...")

        # Train model
        model = RandomForestClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            random_state=random_state
        )
        model.fit(X_train, y_train)

        # Predictions
        y_pred = model.predict(X_test)

        # Metrics
        acc = accuracy_score(y_test, y_pred)
        prec = precision_score(y_test, y_pred)
        rec = recall_score(y_test, y_pred)
        f1 = f1_score(y_test, y_pred)

        logging.info(
            f"Accuracy={acc:.4f}, "
            f"Precision={prec:.4f}, "
            f"Recall={rec:.4f}, "
            f"F1={f1:.4f}"
        )

        # Log parameters
        mlflow.log_param("n_estimators", n_estimators)
        mlflow.log_param("max_depth", max_depth)
        mlflow.log_param("random_state", random_state)

        # Log metrics
        mlflow.log_metric("accuracy", acc)
        mlflow.log_metric("precision", prec)
        mlflow.log_metric("recall", rec)
        mlflow.log_metric("f1_score", f1)

        # Save model locally
        os.makedirs("models", exist_ok=True)
        model_path = os.path.join("models", "model.joblib")
        joblib.dump(model, model_path)
        logging.info(f"Model saved successfully at: {model_path}")

        # Log model artifact to MLflow
        mlflow.log_artifact(model_path, artifact_path="model")
        logging.info("Experiment successfully logged to MLflow.")

    return model, acc


if __name__ == "__main__":
    train_model(n_estimators=100, max_depth=5)

What MLflow Gives Us

Every time train_model() runs, MLflow creates a unique Run ID and stores:

What	Why
Parameters (`n_estimators`, `max_depth`)	So you can compare what configuration produced each result.
Metrics (`accuracy`, `f1_score`)	Track model performance across experiments.
Artifacts (`model.joblib`)	The actual trained binary, versioned and downloadable.

Run it:

python src/train.py
# Output: Accuracy=0.7821, Precision=0.7885, Recall=0.5942, F1=0.6777

# Launch the MLflow dashboard:
python -m mlflow ui
# Open http://localhost:5000 in your browser

You can now visually compare multiple training runs, their hyperparameters, and metrics side-by-side in the MLflow UI.

Step 7: FastAPI Inference Service (`api/main.py`)

End-users don't run Python scripts. They send HTTP requests. Our FastAPI service:

Validates incoming JSON using Pydantic (rejects malformed data with 422 errors).
Transforms the input into the exact feature format the model expects.
Returns a prediction with survival probability.
Records Prometheus metrics for monitoring.

import os
import time
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException, Response
from pydantic import BaseModel, Field
from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST

# 1. Initialize FastAPI app
app = FastAPI(
    title="Titanic Survival Prediction API",
    description="Production-ready inference service with Prometheus monitoring.",
    version="1.0.0"
)

# 2. Prometheus Metrics definitions
REQUEST_COUNT = Counter(
    "api_request_count_total", 
    "Total HTTP requests received", 
    ["method", "endpoint", "status"]
)
REQUEST_LATENCY = Histogram(
    "api_request_latency_seconds", 
    "Histogram of request latencies in seconds", 
    ["endpoint"]
)
PREDICTION_COUNT = Counter(
    "model_prediction_count_total", 
    "Total predictions generated by the model", 
    ["prediction"]
)

# 3. Path to trained model
MODEL_PATH = os.path.join("models", "model.joblib")
if not os.path.exists(MODEL_PATH):
    raise FileNotFoundError(f"Trained model not found at {MODEL_PATH}!")

model = joblib.load(MODEL_PATH)


# 4. Pydantic Schemas
class PassengerInput(BaseModel):
    Pclass: int = Field(..., ge=1, le=3, description="Ticket class (1 = 1st, 2 = 2nd, 3 = 3rd)")
    Sex: str = Field(..., description="Gender: 'male' or 'female'")
    Age: float = Field(..., ge=0, le=120, description="Age in years")
    SibSp: int = Field(..., ge=0, description="Number of siblings/spouses aboard")
    Parch: int = Field(..., ge=0, description="Number of parents/children aboard")
    Fare: float = Field(..., ge=0.0, description="Passenger fare")
    Embarked: str = Field(..., description="Port of Embarkation: 'C', 'Q', or 'S'")


class PredictionResponse(BaseModel):
    survived: bool
    survival_probability: float


# 5. Prometheus Scrape Endpoint
@app.get("/metrics")
def get_metrics():
    """Exposes Prometheus application metrics."""
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)


@app.get("/")
def health_check():
    return {"status": "healthy", "service": "titanic-survival-prediction"}


@app.post("/predict", response_model=PredictionResponse)
def predict(passenger: PassengerInput):
    start_time = time.time()
    try:
        input_data = pd.DataFrame([{
            "Pclass": passenger.Pclass,
            "Age": passenger.Age,
            "SibSp": passenger.SibSp,
            "Parch": passenger.Parch,
            "Fare": passenger.Fare,
            "Sex_male": 1 if passenger.Sex.lower() == "male" else 0,
            "Embarked_Q": 1 if passenger.Embarked.upper() == "Q" else 0,
            "Embarked_S": 1 if passenger.Embarked.upper() == "S" else 0,
        }])

        prediction = model.predict(input_data)[0]
        probability = model.predict_proba(input_data)[0][1]

        # Record metrics
        PREDICTION_COUNT.labels(prediction=str(int(prediction))).inc()
        REQUEST_COUNT.labels(method="POST", endpoint="/predict", status="200").inc()
        REQUEST_LATENCY.labels(endpoint="/predict").observe(time.time() - start_time)

        return PredictionResponse(
            survived=bool(prediction == 1),
            survival_probability=round(float(probability), 4)
        )

    except Exception as e:
        REQUEST_COUNT.labels(method="POST", endpoint="/predict", status="500").inc()
        raise HTTPException(status_code=500, detail=str(e))

The Feature Alignment Trap 🪤

Notice the manual construction of Sex_male, Embarked_Q, and Embarked_S. These must exactly match the dummy columns created during training by pd.get_dummies(..., drop_first=True). If you forget Embarked_Q or add Sex_female instead of Sex_male, the model will silently produce garbage predictions — no errors, just wrong numbers.

API Example

Request:

{
  "Pclass": 1,
  "Sex": "female",
  "Age": 29.0,
  "SibSp": 0,
  "Parch": 0,
  "Fare": 100.0,
  "Embarked": "S"
}

Response:

{
  "survived": true,
  "survival_probability": 0.9200
}

Run it:

uvicorn api.main:app --reload --port 8000
# Open http://localhost:8000/docs for interactive Swagger UI

Step 8: Automated Testing with Pytest

Tests are CI/CD guardrails. If any test fails, deployment stops automatically.

`tests/test_model.py` — Model Integrity Tests

import os
import joblib
import pandas as pd


def test_model_file_exists():
    """Verify the trained model artifact exists on disk."""
    model_path = os.path.join("models", "model.joblib")
    assert os.path.exists(model_path), "Model artifact 'model.joblib' is missing!"


def test_model_prediction_output():
    """Verify model can take a sample input row and output 0 or 1."""
    model_path = os.path.join("models", "model.joblib")
    model = joblib.load(model_path)

    sample = pd.DataFrame([{
        "Pclass": 3,
        "Age": 22.0,
        "SibSp": 1,
        "Parch": 0,
        "Fare": 7.25,
        "Sex_male": 1,
        "Embarked_Q": 0,
        "Embarked_S": 1
    }])

    pred = model.predict(sample)
    assert pred[0] in [0, 1], f"Unexpected prediction value: {pred[0]}"

`tests/test_api.py` — API Contract Tests

from fastapi.testclient import TestClient
from api.main import app

client = TestClient(app)


def test_health_check():
    """Test GET / returns healthy status."""
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"status": "healthy", "service": "titanic-survival-prediction"}


def test_predict_endpoint_valid_input():
    """Test POST /predict with valid passenger JSON."""
    payload = {
        "Pclass": 1,
        "Sex": "female",
        "Age": 29.0,
        "SibSp": 0,
        "Parch": 0,
        "Fare": 100.0,
        "Embarked": "S"
    }
    response = client.post("/predict", json=payload)
    assert response.status_code == 200

    data = response.json()
    assert "survived" in data
    assert "survival_probability" in data
    assert isinstance(data["survived"], bool)
    assert 0.0 <= data["survival_probability"] <= 1.0


def test_predict_endpoint_invalid_input():
    """Test POST /predict fails when invalid data is passed."""
    invalid_payload = {
        "Pclass": 1,
        "Sex": "female",
        "Age": -5.0,  # Negative age should be rejected by Pydantic ge=0
        "SibSp": 0,
        "Parch": 0,
        "Fare": 50.0,
        "Embarked": "S"
    }
    response = client.post("/predict", json=invalid_payload)
    assert response.status_code == 422  # Unprocessable Entity

Run:

pytest tests/ -v
# tests/test_model.py::test_model_file_exists PASSED
# tests/test_model.py::test_model_prediction_output PASSED
# tests/test_api.py::test_health_check PASSED
# tests/test_api.py::test_predict_endpoint_valid_input PASSED
# tests/test_api.py::test_predict_endpoint_invalid_input PASSED

Step 9: Docker Containerization

Why Docker?

Your model works on your machine with Python 3.13 on Windows. But it might fail on an Ubuntu cloud server because of missing C++ build tools, conflicting package versions, or different file paths. Docker packages your Python version, dependencies, model artifact, and FastAPI server into an isolated, reproducible Linux container.

`Dockerfile`

# 1. Use an official lightweight Python runtime
FROM python:3.11-slim

# 2. Set environment variables
ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

# 3. Set working directory inside container
WORKDIR /app

# 4. Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 5. Copy necessary application code and models
COPY api/ ./api/
COPY src/ ./src/
COPY models/ ./models/

# 6. Expose the port FastAPI runs on
EXPOSE 8000

# 7. Command to run the application using Uvicorn
CMD ["uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000"]

Layer Caching Trick 🚀

Notice that COPY requirements.txt and RUN pip install come before COPY api/ and COPY src/. Docker caches layers. If you only change your source code (not dependencies), Docker skips the entire pip install step and rebuilds in ~2 seconds instead of 2 minutes.

`.dockerignore`

venv/
.env
__pycache__/
*.pyc
.git/
.dvc/
data/raw/
mlruns/
mlflow.db
tests/
notebooks/

`docker-compose.yml`

services:
  titanic-api:
    image: titanic-mlops
    container_name: titanic-mlops-api
    ports:
      - "8000:8000"
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    container_name: titanic-prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml
    restart: unless-stopped

Build & Run

# Build the image
docker build -t titanic-mlops .

# Run with Docker Compose (API + Prometheus)
docker compose up -d

Now visit:

API: http://localhost:8000/docs
Prometheus: http://localhost:9090

Step 10: CI/CD with GitHub Actions

Every push to main triggers an automated pipeline that validates data, trains the model, runs tests, and builds the Docker image — all on GitHub's free cloud runners.

`.github/workflows/ci_cd.yml`

name: Titanic MLOps Pipeline

on:
  push:
    branches: [ "main", "master" ]
  pull_request:
    branches: [ "main", "master" ]

jobs:
  mlops_pipeline:
    runs-on: ubuntu-latest

    steps:
    - name: Checkout Code
      uses: actions/checkout@v4

    - name: Set up Python 3.11
      uses: actions/setup-python@v5
      with:
        python-version: "3.11"
        cache: 'pip'

    - name: Install Dependencies
      run: |
        python -m pip install --upgrade pip
        pip install -r requirements.txt

    - name: Run Data Validation
      run: |
        python src/data_validation.py

    - name: Train Model
      run: |
        python src/train.py

    - name: Run Pytest Suite
      env:
        PYTHONPATH: .
      run: |
        pytest tests/ -v

    - name: Build Docker Image
      run: |
        docker build -t titanic-mlops-api:latest .

The `PYTHONPATH: .` Fix

On Linux CI runners, the project root isn't automatically in Python's import search path. Without PYTHONPATH: ., pytest can't find the api module and fails with ModuleNotFoundError: No module named 'api'.

Push and watch it run:

git add .
git commit -m "Add CI/CD pipeline"
git push origin main

Check the Actions tab on your GitHub repository to see the pipeline execute live.

Step 11: Production Monitoring with Prometheus

What does Prometheus track?

In our api/main.py, we defined three metric types:

Metric	Type	What It Measures
`api_request_count_total`	Counter	Total HTTP requests by endpoint and status code
`api_request_latency_seconds`	Histogram	Inference response time in seconds
`model_prediction_count_total`	Counter	How many survived (1) vs died (0) predictions

`monitoring/prometheus.yml`

global:
  scrape_interval: 5s

scrape_configs:
  - job_name: "titanic-api"
    metrics_path: "/metrics"
    static_configs:
      - targets: ["titanic-api:8000"]

Prometheus pulls from the FastAPI /metrics endpoint every 5 seconds and stores time-series data.

Why This Matters in Production

Imagine your model is deployed and serving 10,000 requests per day. Prometheus helps you detect:

Traffic spikes: Request count suddenly 10x → is this organic growth or a bot attack?
Latency degradation: Average inference time jumps from 15ms to 800ms → your server is out of memory.
Prediction distribution shift: Suddenly 99% of predictions are survived=0 when it's normally 38% → something is wrong with the incoming data.

Querying Prometheus

After running some predictions through the API, visit http://localhost:9090 and query:

api_request_count_total
model_prediction_count_total
rate(api_request_latency_seconds_sum[5m]) / rate(api_request_latency_seconds_count[5m])

Step 12: Data Drift Detection with Evidently AI

What is Data Drift?

A model trained on 1912 Titanic passengers assumes certain demographic distributions — median age ~28, mostly 3rd class, mostly male. If the model is deployed live and incoming traffic suddenly changes (e.g., all passengers are age 60+ with fares >$300), the model won't crash — it will silently produce unreliable, inaccurate predictions.

This is called Data Drift (or Covariate Shift), and it's the #1 silent killer of production ML systems.

`src/drift_detection.py`

import os
import sys
import logging
import pandas as pd

# Support both new and older Evidently versions
try:
    from evidently.report import Report
    from evidently.metric_preset import DataDriftPreset, DataQualityPreset
except ModuleNotFoundError:
    from evidently.legacy.report import Report
    from evidently.legacy.metric_preset import DataDriftPreset, DataQualityPreset

sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
from src.data_ingestion import load_data
from src.data_preprocess import preprocess_data

logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")


def generate_drift_report(
    reference_data: pd.DataFrame, 
    current_data: pd.DataFrame, 
    output_html_path: str = "monitoring/drift_report.html"
):
    """
    Compares reference (training) data with current (production) data
    and generates an interactive HTML drift dashboard.
    """
    logging.info("Building Evidently Data Drift and Data Quality Report...")

    features = [col for col in reference_data.columns if col != "Survived"]
    ref_df = reference_data[features]
    curr_df = current_data[features]

    report = Report(metrics=[
        DataDriftPreset(),
        DataQualityPreset()
    ])

    report.run(reference_data=ref_df, current_data=curr_df)

    os.makedirs(os.path.dirname(output_html_path), exist_ok=True)
    report.save_html(output_html_path)

    logging.info(f"Drift report successfully generated at: {output_html_path}")
    return output_html_path


if __name__ == "__main__":
    # Load baseline reference data
    raw_path = os.path.join("data", "raw", "titanic.csv")
    df = load_data(raw_path)
    reference_df = preprocess_data(df)

    # Simulate "Current" production data with intentional drift!
    current_df = reference_df.copy()
    current_df["Age"] = current_df["Age"] + 25.0  # Simulated drift in Age
    current_df["Fare"] = current_df["Fare"] * 3.5  # Simulated drift in Fare

    # Generate drift report
    report_file = generate_drift_report(reference_df, current_df)
    print(f"\nReport ready! Open in browser: {os.path.abspath(report_file)}")

How Evidently Detects Drift

Evidently runs statistical tests on every feature:

Feature Type	Statistical Test	What It Measures
Numerical (`Age`, `Fare`)	Kolmogorov-Smirnov test	Whether two distributions are from the same population
Categorical (`Pclass`, `Sex_male`)	Chi-Square test	Whether category proportions have changed

If the p-value falls below the threshold (default 0.05), the feature is flagged as drifted.

Run it:

python src/drift_detection.py
# Drift report successfully generated at: monitoring/drift_report.html

Open monitoring/drift_report.html in your browser to see the interactive Evidently dashboard with per-feature distribution charts, p-values, and drift flags.

The Complete Picture

Let's step back and appreciate what we've built:

┌──────────────────────────────────────────────────────┐
│                  Developer Pushes Code               │
└────────────────────────┬─────────────────────────────┘
                         │
                         ▼
┌──────────────────────────────────────────────────────┐
│              GitHub Actions CI/CD Pipeline            │
│                                                      │
│  ✅ Install Dependencies                             │
│  ✅ Validate Data Schema                             │
│  ✅ Train Model + Log to MLflow                      │
│  ✅ Run Pytest Suite (Model + API tests)             │
│  ✅ Build Docker Image                               │
└────────────────────────┬─────────────────────────────┘
                         │
                         ▼
┌──────────────────────────────────────────────────────┐
│            Docker Container (Production)             │
│                                                      │
│  FastAPI (:8000)                                     │
│  ├── POST /predict  → Model Inference                │
│  ├── GET  /metrics  → Prometheus Scrape Endpoint     │
│  └── GET  /docs     → Interactive Swagger UI         │
└────────────────────────┬─────────────────────────────┘
                         │
              ┌──────────┴──────────┐
              ▼                     ▼
┌────────────────────┐  ┌────────────────────┐
│  Prometheus (:9090) │  │  Evidently AI       │
│                    │  │                    │
│  Request Counts    │  │  Data Drift        │
│  Latency Tracking  │  │  Feature Drift     │
│  Prediction Dist.  │  │  Quality Metrics   │
└────────────────────┘  └────────────────────┘

Key Takeaways

Data Versioning (DVC): Never commit raw data to Git. Track it with DVC for reproducibility and rollback.
Validate Before You Train: Schema checks, constraint checks, and missing-value guardrails catch data problems before they silently corrupt your model.
Track Every Experiment (MLflow): Every training run should log parameters, metrics, and artifacts. "I think I got 82% accuracy last Tuesday" is not acceptable in production.
Serve via API, Not Scripts: End-users send HTTP requests. FastAPI with Pydantic validation gives you type-safe, self-documenting endpoints.
Test Everything Automatically: Model integrity tests + API contract tests in Pytest, executed automatically in CI.
Containerize for Reproducibility: Docker eliminates "works on my machine" problems entirely.
Automate with CI/CD: GitHub Actions ensures every push is validated, tested, trained, and container-built before reaching production.
Monitor in Production: Prometheus gives you real-time visibility into traffic, latency, and prediction behavior. Without it, you're flying blind.
Detect Drift: Evidently AI catches the silent killer of ML systems — when production data distribution shifts away from what the model was trained on.

Future Enhancements

Automated Retraining Pipeline: Trigger model retraining when Evidently detects significant drift.
Feature Store Integration: Centralized feature management with Feast or Hopsworks.
A/B Testing: Serve two model versions simultaneously and compare live performance.
Kubernetes Deployment: Scale from single Docker container to orchestrated cluster.
Grafana Dashboards: Beautiful visualization layer on top of Prometheus metrics.
Real-time Streaming: Replace batch inference with real-time event-driven predictions.

Resources

GitHub Repo: End-to-end-MLOPS-Pipeline
DVC Documentation
MLflow Documentation
FastAPI Documentation
Evidently AI Documentation
Prometheus Documentation

If you found this walkthrough helpful, consider giving the GitHub repo a ⭐ and sharing this post with someone learning MLOps!

Building a Production-Grade End-to-End MLOps Pipeline from Scratch

Architecture Overview

Technology Stack

Project Structure

Step 1: Project Initialization & Environment Setup

.gitignore

requirements.txt

Step 2: Data Versioning with DVC

Why not just commit datasets to Git?

Setup

Step 3: Data Ingestion (src/data_ingestion.py)

Why logging instead of print()?

Step 4: Data Validation (src/data_validation.py)

What are we defending against?

Step 5: Preprocessing & Feature Engineering (src/data_preprocess.py)

Key Decisions Explained

Step 6: Model Training & Experiment Tracking with MLflow (src/train.py)

What MLflow Gives Us

Step 7: FastAPI Inference Service (api/main.py)

The Feature Alignment Trap 🪤

API Example

Step 8: Automated Testing with Pytest

tests/test_model.py — Model Integrity Tests

tests/test_api.py — API Contract Tests

Step 9: Docker Containerization

Why Docker?

Dockerfile

Layer Caching Trick 🚀

.dockerignore

docker-compose.yml

Build & Run

Step 10: CI/CD with GitHub Actions

.github/workflows/ci_cd.yml

The PYTHONPATH: . Fix

Step 11: Production Monitoring with Prometheus

What does Prometheus track?

monitoring/prometheus.yml

Why This Matters in Production

Querying Prometheus

Step 12: Data Drift Detection with Evidently AI

What is Data Drift?

src/drift_detection.py

How Evidently Detects Drift

The Complete Picture

Key Takeaways

Future Enhancements

Resources

`.gitignore`

`requirements.txt`

Step 3: Data Ingestion (`src/data_ingestion.py`)

Why `logging` instead of `print()`?

Step 4: Data Validation (`src/data_validation.py`)

Step 5: Preprocessing & Feature Engineering (`src/data_preprocess.py`)

Step 6: Model Training & Experiment Tracking with MLflow (`src/train.py`)

Step 7: FastAPI Inference Service (`api/main.py`)

`tests/test_model.py` — Model Integrity Tests

`tests/test_api.py` — API Contract Tests

`Dockerfile`

`.dockerignore`

`docker-compose.yml`

`.github/workflows/ci_cd.yml`

The `PYTHONPATH: .` Fix

`monitoring/prometheus.yml`

`src/drift_detection.py`