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

> Source: <https://dev.to/naman_2004/building-a-production-grade-end-to-end-mlops-pipeline-from-scratch-l9h>
> Published: 2026-09-18 18:21:06+00:00

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](https://github.com/naman-0804/End-to-end-MLOPS-Pipeline)

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)
```

| 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 | 

```
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/
```

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
```

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:

`.dvc` metadata files in Git (a few bytes).`git checkout`.
Download the [Titanic dataset](https://raw.githubusercontent.com/datasciencedojo/datasets/master/titanic.csv) 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.

`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.

``` python
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())
```

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

`INFO`, `WARNING`, `ERROR`) to filter noise.` 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.

``` python
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)
```

| 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!
```

`src/data_preprocess.py`)

``` python
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)}")
```

`fillna` with median/mode`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)`` Sex` (male/female) → `Sex_male` (1/0) and `Embarked` (C/Q/S) → `Embarked_Q`, `Embarked_S`. The `drop_first` avoids the `stratify=y`

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

`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.

``` python
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)
```

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. | 

```
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.

`api/main.py`)
End-users don't run Python scripts. They send HTTP requests. Our FastAPI service:

``` python
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))
```

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.

**Request:**

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

**Response:**

```
{
  "survived": true,
  "survival_probability": 0.9200
}
uvicorn api.main:app --reload --port 8000
# Open http://localhost:8000/docs for interactive Swagger UI
```

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

`tests/test_model.py` — Model Integrity Tests

``` python
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

``` python
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
```

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"]
```

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 the image
docker build -t titanic-mlops .

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

Now visit:

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 .
```

`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.

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. 

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

`survived=0` when it's normally 38% → something is wrong with the incoming data.
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])
```

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`

``` python
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)}")
```

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**.

```
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.

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   │
└────────────────────┘  └────────────────────┘
```

**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.

*If you found this walkthrough helpful, consider giving the [GitHub repo](https://github.com/naman-0804/End-to-end-MLOPS-Pipeline) a ⭐ and sharing this post with someone learning MLOps!*