The previous chapter got a single project deployed. MLOps is the discipline of doing that repeatedly and reliably: versioning data and models, tracking every experiment, automating tests, and shipping updates through CI/CD like any professional software team. Where a one-off project ends at deployment, MLOps begins there.
Every MLOps project starts with a reproducible environment and a clean folder layout. Create a project folder, add a virtual environment, and install the core tooling:
python -m venv venv
venv\Scripts\activate
pip install numpy pandas scikit-learn pymongo dvc mlflow boto3 dillAdopt a layout where src holds source code, config holds configuration files, data holds inputs and outputs, and artifacts holds trained models. A standard structure keeps every team member able to find any file without asking.
.gitignore entries for venv/, __pycache__/, data files, and environment secrets.Use VS Code's integrated source control panel (the branch icon) to stage and commit instead of the terminal if you prefer; the commands underneath are the same git add and git commit from the previous chapter.
A setup.py turns your source folder into a proper Python package, so modules can import each other with plain names like from src.exception import CustomException instead of fragile relative paths.
from setuptools import setup, find_packages
from typing import List
def get_requirements() -> List[str]:
with open("requirements.txt") as f:
return [r for r in f.read().splitlines() if r and not r.startswith("-e .")]
setup(
name="mlops-project",
version="0.1.0",
packages=find_packages(),
install_requires=get_requirements(),
)Add -e . to requirements.txt if you want the package installed in editable mode, so every change to the source is picked up immediately during development.
An MLOps pipeline runs unattended, often at odd hours, so it must leave a clear trail. Centralise logging in one module every other file imports:
import logging, os
os.makedirs("logs", exist_ok=True)
logging.basicConfig(
filename="logs/pipeline.log",
level=logging.INFO,
format="%(asctime)s %(levelname)s %(module)s: %(message)s",
)
def get_logger(name):
return logging.getLogger(name)Pair it with a custom exception class that logs the full traceback when raised, and wrap every component call so a failure names the exact stage that broke:
class CustomException(Exception):
def __init__(self, message):
self.message = message
super().__init__(message)
logger.error(f"{message} | {sys.exc_info()}")ETL stands for Extract, Transform, Load. It is the classic recipe for moving data from where it is produced to where analysis happens:
Many ML projects run a smaller version of this: extract a raw dataset, transform it with preprocessing, and load it into a store the training pipeline reads from.
MongoDB Atlas is a hosted NoSQL document database that stores data as flexible JSON-like documents. It is a convenient destination for ETL output because documents do not need a fixed schema.
mongodb+srv://user:password@cluster.mongodb.net/.Here is a small ETL script that reads a CSV, cleans a column, and loads the result into MongoDB:
import pandas as pd
from pymongo import MongoClient
def extract(path="data/raw.csv"):
return pd.read_csv(path)
def transform(df):
df = df.dropna(subset=["scores"])
df["scores"] = df["scores"].astype(float)
return df
def load(df):
client = MongoClient(os.environ["MONGO_URI"])
db = client["mlops"]
db["students"].insert_many(df.to_dict("records"))
load(transform(extract()))Run the stages in the right order and log the row counts before and after each step, so the pipeline reports how much data actually moved.
Data ingestion is the component that loads data from the ETL destination into the training workflow. Architect it in three layers:
Separating configuration from behaviour means the same code runs against different datasets by only changing configuration, which is exactly what makes pipelines reusable.
Hold the settings in a dataclass so the ingestion component never hardcodes paths:
@dataclass
class DataIngestionConfig:
train_path: str = os.path.join("artifacts", "train.csv")
test_path: str = os.path.join("artifacts", "test.csv")
raw_path: str = os.path.join("artifacts", "data.csv")The component reads from MongoDB (or another source), performs a reproducible split, and saves the split files:
class DataIngestion:
def __init__(self, config: DataIngestionConfig):
self.config = config
def initiate_ingestion(self):
df = load_from_mongodb("mlops", "students")
train_df = df.sample(frac=0.8, random_state=42)
test_df = df.drop(train_df.index)
train_df.to_csv(self.config.train_path, index=False)
test_df.to_csv(self.config.test_path, index=False)
return self.config.train_path, self.config.test_pathModels are only as good as the data they see, so the pipeline must verify the data before spending compute on training.
Validation first checks that expected columns exist, are of the right types, and have acceptable ranges or value sets. A second check is data drift: compare the distribution of the new batch against the reference data the model was trained on. Large drift is a warning that the model may be making decisions in a world it never saw.
expected = {"scores": "float64", "study_hours": "int64"}
for col, dtype in expected.items():
if col not in df or str(df[col].dtype) != dtype:
raise CustomException(f"Validation failed on {col}")Only after checks pass do we write validated train and test files to the artifacts folder and log the outcome. Reject the run on failure instead of silently training on bad data.
Transformation converts validated raw data into the numeric feature matrix a model can consume.
Use scikit-learn's ColumnTransformer with separate pipelines for numeric and categorical columns, and save the fitted transformer as an artifact. The same object must be reused at prediction time so new data is transformed identically to training data.
num_pipeline = Pipeline([
("imputer", SimpleImputer(strategy="median")),
("scaler", StandardScaler()),
])
cat_pipeline = Pipeline([
("imputer", SimpleImputer(strategy="most_frequent")),
("onehot", OneHotEncoder(handle_unknown="ignore")),
])
preprocessor = ColumnTransformer([
("num", num_pipeline, numeric_features),
("cat", cat_pipeline, categorical_features),
])Fit on the training split only, transform both splits, and persist the preprocessor with joblib.dump.
The trainer fits a candidate model on the transformed training data and reports a baseline metric on the held-out set. Part 1 is deliberately simple: one algorithm, fixed settings, one score, all logged:
class ModelTrainer:
def __init__(self, model_path="artifacts/model.pkl"):
self.model_path = model_path
def train(self, X_train, X_test, y_train, y_test):
model = RandomForestRegressor(random_state=42)
model.fit(X_train, y_train)
r2 = model.score(X_test, y_test)
logger.info(f"Baseline R2 = {r2}")
return modelNow evaluate honestly with several metrics and tune the hyperparameters with cross-validated search. Log every metric, not just the headline one:
from sklearn.metrics import r2_score, mean_absolute_error
from sklearn.model_selection import RandomizedSearchCV
params = {"n_estimators": [100, 300, 500], "max_depth": [5, 10, None]}
search = RandomizedSearchCV(RandomForestRegressor(), params,
cv=5, n_iter=10, random_state=42)
search.fit(X_train, y_train)
best = search.best_estimator_
preds = best.predict(X_test)
logger.info(f"R2={r2_score(y_test, preds):.3f} "
f"MAE={mean_absolute_error(y_test, preds):.3f}")MLflow records experiments so you can compare runs later. Each run logs parameters, metrics, and artifacts to a central store:
import mlflow
with mlflow.start_run(run_name="rf-tuned"):
mlflow.log_params(search.best_params_)
mlflow.log_metric("r2", r2)
mlflow.log_metric("mae", mae)
mlflow.log_artifact("artifacts/model.pkl")With MLflow you can look back over hundreds of runs and answer the question every ML team eventually asks: "which experiment actually produced the best model, and with what settings?"
Local MLflow files help one person. A DagsHub repository hosts the experiment store in the cloud so a team can share it. DagsHub is a hosting platform for data science projects that integrates Git, DVC, and MLflow together.
MLFLOW_TRACKING_URI.mlflow.set_tracking_uri(os.environ["MLFLOW_TRACKING_URI"]).Tracking and experimenting produce many candidate models, but only one should reach production. The model pusher selects the best registered model and moves it into the production path, a clearly named file the deployment reads:
class ModelPusher:
def __init__(self, best_model_path, prod_path):
self.best_model_path = best_model_path
self.prod_path = prod_path
def push(self):
shutil.copy(self.best_model_path, self.prod_path)
logger.info(f"Pushed model to {self.prod_path}")
return self.prod_pathKeeping one production path makes the serving code trivial and lets a rollback be a simple copy of the previous artifact.
The training pipeline is a single entry point that calls every component in order: ingestion, validation, transformation, training, tracking, and pushing. The batch prediction pipeline reads new data, applies the saved preprocessor, loads the production model, and writes predictions back to a destination such as a CSV or database:
def run_batch_predictions(new_data_path, out_path):
df = pd.read_csv(new_data_path)
features = preprocessor.transform(df)
preds = prod_model.predict(features)
df["prediction"] = preds
df.to_csv(out_path, index=False)Batch prediction suits cases where answers are not needed instantly: nightly scoring of loan applications, churn scores refreshed each morning, and so on.
Models and data should outlive the machine that trained them. S3 is AWS object storage and the natural home for final artifacts. Upload with the AWS SDK:
import boto3
s3 = boto3.client("s3")
s3.upload_file("artifacts/model.pkl", "my-ml-bucket", "models/model.pkl")
s3.upload_file("artifacts/preprocessor.pkl", "my-ml-bucket", "models/preprocessor.pkl")Store training data, transformed data, and models under clearly named keys, and version them by date. A new machine can now reconstruct the entire pipeline from S3 alone.
The deployment pipeline builds a Docker image of the serving app, then runs it in the cloud. GitHub Actions automates the whole process: every push to the main branch triggers a workflow that builds the image and ships it. A workflow file lives in .github/workflows/main.yml:
name: build-and-deploy
on:
push:
branches: [main]
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Configure AWS credentials
uses: aws-actions/configure-aws-credentials@v4
with:
aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
aws-region: us-east-1Never write real keys in the file; reference them as repository secrets.
Continuing the workflow, the job builds the Docker image and pushes it to Amazon ECR. The pattern mirrors the manual steps from the previous chapter, automated:
- name: Login to Amazon ECR
run: |
aws ecr get-login-password --region us-east-1 | \
docker login --username AWS --password-stdin \
${{ secrets.ECR_REGISTRY }}
- name: Build and push image
run: |
docker build -t ${{ secrets.ECR_REGISTRY }}/my-app:latest .
docker push ${{ secrets.ECR_REGISTRY }}/my-app:latestAfter this step completes, a fresh image of your app is sitting in the registry, waiting to be pulled by the server that will host it.
docker pull 123456789.dkr.ecr.us-east-1.amazonaws.com/my-app:latest
docker run -d -p 8080:8080 \
-e MONGO_URI=... 123456789.dkr.ecr.us-east-1.amazonaws.com/my-app:latestThe app is now live, reproducible from the image, and updateable by simply pushing a new image and restarting the container.
If MLOps feels overwhelming, start small: use MLflow locally on your next single experiment. Log three things per run, parameters, metrics, and the model file, then open the UI with mlflow ui. Once that habit is established, move the tracking server to DagsHub so the whole team sees the same experiments. Incremental adoption beats a big-bang rewrite every time.
Models are meaningless if you cannot reproduce the data they trained on. DVC (Data Version Control) brings Git-style versioning to datasets and models, which live outside Git because they are too large:
dvc init
dvc remote add -d myremote s3://my-ml-bucket/dvc-store
dvc add data/train.csv
git add data/train.csv.dvc .dvc
git commit -m "track dataset"
dvc pushA tiny pointer file records the data hash in Git, while the real file lives in the remote. Checking out an old commit also restores the exact data version, so every experiment is reproducible.
BentoML is a modern serving framework that packages a trained model into a bento: a self-contained bundle with the model, the Python environment, and an HTTP API. Building a bento is declarative:
import bentoml
from bentoml.io import JSON, NumpyNdarray
runner = bentoml.sklearn.get("my_model:latest").to_runner()
svc = bentoml.Service("student-predictor", runners=[runner])
@svc.api(input=JSON(), output=JSON())
def predict(input_data):
features = input_data["features"]
result = runner.predict.run([features])
return {"prediction": float(result[0])}Then build the bento with bentoml build and serve it with bentoml serve. BentoML handles versioning, packaging, and even converting the bento into a Docker image or serverless function, which makes it a strong final piece of the modern MLOps toolbox.