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

A developer published a walkthrough for building a production-grade end-to-end MLOps pipeline around the Titanic survival prediction dataset, covering data versioning with DVC, validation, scikit-learn training, MLflow experiment tracking, Pytest testing, Docker containerization, GitHub Actions CI/CD, FastAPI serving, Prometheus monitoring, and Evidently AI drift detection. The project, released on GitHub as End-to-end-MLOPS-Pipeline, emphasizes that model training is a small fraction of production ML work.

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