MLOps · Lesson 58 of 80

Capstone: The Plan

A complete project, step by step.

⏱ 2 min read▶ Video at 2:26:28🎓 Free

Watch this lesson in the MLOps courseWatch on YouTube ↗Full MLOps course

Now it is your turn to see everything working together. In this capstone, we build a complete churn prediction system, step by step, and every command you will see is real. We use only what you have learned. Data versioning, validation, a pipeline, tracking, a registry with a quality gate, serving, deployment, and monitoring.

One project, every skill
Every command is real
From data to monitoring

The capstone plan

Here is the plan, in six steps. One, set up the repository, and generate and validate the data. Two, build a pipeline with three stages, prepare, validate, and train, with tracking. Three, test the model, and promote it through a quality gate into the registry. Four, package it, and serve it, in a container. Five, describe the deployment, and automate it with a pipeline. And six, monitor it, detect drift, and retrain. Each step produces something you can inspect.

Step	What we build
1 Repository	layout, data, validation
2 Pipeline	prepare, validate, train, tracked
3 Quality gate	tests, registry, champion alias
4 Serving	container and API
5 Delivery	Kubernetes and CI/CD
6 Operate	monitor, drift, retrain

Step one: the repository

Step one. We write the project files, which you will see one by one. We start a Git repository, and a D V C project. Here is the layout, a source folder with scripts for preparing, validating, training, promoting, and checking drift, plus the pipeline definition, the parameters, the tests, and the files for serving. Then we generate the data, and validate it against the contract. Four thousand rows, and the data is valid.

terminal

git init -q -b main && dvc init -q
find . -type f -not -path './.git/*' -not -path './.dvc/*' | sort | head -16
./.dvcignore
./Dockerfile
./app.py
./dvc.yaml
./params.yaml
./requirements.txt
./src/churn.py
./src/drift.py
./src/incoming.py
./src/prepare.py
./src/promote.py
./src/train.py
./src/validate.py
./tests/test_quality.py
python src/prepare.py 1
prepared data/churn.csv, seed 1 normal
python src/validate.py
data valid: 4000 rows

File used: src/prepare.py

import yaml
from churn import make_data

params = yaml.safe_load(open("params.yaml"))["prepare"]
make_data(4000, seed=params["seed"], shifted=params["shifted"]).to_csv("data/churn.csv", index=False)
print("prepared data/churn.csv, seed", params["seed"], "shifted" if params["shifted"] else "normal")

File used: src/validate.py

import pandas as pd
import pandera.pandas as pa

schema = pa.DataFrameSchema({
    "tenure": pa.Column(int, pa.Check.in_range(0, 100)),
    "charges": pa.Column(float, pa.Check.gt(0)),
    "calls": pa.Column(int, pa.Check.ge(0)),
    "contract": pa.Column(int, pa.Check.isin([0, 1, 2])),
    "churn": pa.Column(int, pa.Check.isin([0, 1])),
})
df = pd.read_csv("data/churn.csv")
schema.validate(df)
print("data valid:", len(df), "rows")

File used: src/train.py

import json
import joblib
import mlflow
import pandas as pd
import yaml
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score
from sklearn.model_selection import train_test_split

params = yaml.safe_load(open("params.yaml"))["train"]
df = pd.read_csv("data/churn.csv")
X, y = df.drop(columns="churn"), df["churn"]
Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.25, random_state=0, stratify=y)

mlflow.set_tracking_uri("sqlite:///mlflow.db")
mlflow.set_experiment("churn-capstone")
with mlflow.start_run():
    model = RandomForestClassifier(**params, class_weight="balanced", random_state=0).fit(Xtr, ytr)
    f1 = f1_score(yte, model.predict(Xte))
    mlflow.log_params(params)
    mlflow.log_metric("f1", f1)

joblib.dump(model, "models/model.joblib")
json.dump({"f1": round(f1, 3)}, open("metrics.json", "w"))
print("f1", round(f1, 3))

File used: params.yaml

prepare:
  seed: 1
  shifted: false
train:
  n_estimators: 100
  max_depth: 6

File used: dvc.yaml

stages:
  prepare:
    cmd: python src/prepare.py
    deps:
    - src/prepare.py
    params:
    - prepare
    outs:
    - data/churn.csv
  validate:
    cmd: python src/validate.py
    deps:
    - src/validate.py
    - data/churn.csv
  train:
    cmd: python src/train.py
    deps:
    - src/train.py
    - data/churn.csv
    params:
    - train
    outs:
    - models/model.joblib
    metrics:
    - metrics.json:
        cache: false

File used: src/promote.py

import json
import joblib
import mlflow
from mlflow import MlflowClient

mlflow.set_tracking_uri("sqlite:///mlflow.db")
client = MlflowClient()
NAME = "churn-model"

candidate_f1 = json.load(open("metrics.json"))["f1"]
model = joblib.load("models/model.joblib")

try:
    champion = client.get_model_version_by_alias(NAME, "champion")
    champion_f1 = float(champion.tags["f1"])
except Exception:
    champion_f1 = 0.30

if candidate_f1 > champion_f1:
    with mlflow.start_run():
        info = mlflow.sklearn.log_model(model, name="model", serialization_format="cloudpickle")
    version = mlflow.register_model(info.model_uri, NAME)
    client.set_model_version_tag(NAME, version.version, "f1", str(candidate_f1))
    client.set_registered_model_alias(NAME, "champion", version.version)
    print(f"PROMOTED version {version.version}: f1 {candidate_f1} beats {champion_f1}")
else:
    print(f"REJECTED: f1 {candidate_f1} does not beat champion {champion_f1}")

File used: tests/test_quality.py

import json
import joblib
import pandas as pd

def test_model_clears_the_quality_floor():
    assert json.load(open("metrics.json"))["f1"] >= 0.30

def test_model_returns_valid_probabilities():
    model = joblib.load("models/model.joblib")
    row = pd.DataFrame([{"tenure": 3, "charges": 95.0, "calls": 4, "contract": 0}])
    probability = model.predict_proba(row)[0][1]
    assert 0.0 <= probability <= 1.0

File used: src/churn.py

import numpy as np, pandas as pd

def make_data(n, seed, shifted=False):
    rng = np.random.default_rng(seed)
    tenure = rng.integers(1, 72, n)
    charges = rng.normal(65 + (28 if shifted else 0), 18, n).clip(20, 140).round(2)
    calls = rng.poisson(1.5 + (1.5 if shifted else 0), n)
    contract = rng.choice([0, 1, 2], n, p=[.55, .25, .20] if not shifted else [.30, .30, .40])
    logit = -1.0 - 0.04 * tenure + 0.025 * (charges - 65) + 0.45 * calls - 0.8 * contract
    if shifted:
        logit = -0.2 - 0.01 * tenure - 0.01 * (charges - 65) + 0.2 * calls - 0.2 * contract
    churn = (rng.random(n) < 1 / (1 + np.exp(-logit))).astype(int)
    return pd.DataFrame({"tenure": tenure, "charges": charges, "calls": calls, "contract": contract, "churn": churn})

File used: src/incoming.py

from churn import make_data

make_data(1500, seed=9, shifted=True).to_csv("data/incoming.csv", index=False)
print("wrote data/incoming.csv (this week's data)")

File used: src/drift.py

import sys
import pandas as pd
from scipy.stats import ks_2samp

reference = pd.read_csv("data/churn.csv")
incoming = pd.read_csv("data/incoming.csv")

drifted = []
for column in ["tenure", "charges", "calls"]:
    p_value = ks_2samp(reference[column], incoming[column]).pvalue
    status = "DRIFT" if p_value < 0.01 else "ok"
    if status == "DRIFT":
        drifted.append(column)
    print(f"{column:8s} p-value {p_value:.4f}  {status}")

if drifted:
    print("drift in", drifted, "-> retrain")
    sys.exit(1)
print("no drift")

File used: app.py

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

model = joblib.load("models/model.joblib")
app = FastAPI(title="Churn API")


class Customer(BaseModel):
    tenure: int
    charges: float
    calls: int
    contract: int


@app.get("/health")
def health():
    return {"status": "ok"}


@app.post("/predict")
def predict(customer: Customer):
    row = pd.DataFrame([customer.model_dump()])
    probability = float(model.predict_proba(row)[0][1])
    return {"churn_probability": round(probability, 3), "model_version": "1"}

File used: requirements.txt

fastapi==0.162.0
joblib==1.6.0
numpy==2.5.4
pandas==3.1.0
pydantic==2.14.2
scikit-learn==1.9.1
uvicorn==0.54.0

File used: Dockerfile

FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY app.py .
COPY models/model.joblib models/model.joblib
EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

Summary

Good. We have a plan with six steps, a clean repository, and validated data. Next, step two, the pipeline, with tracking, so that one command reproduces everything.

Key takeaways

Six steps, all real
Repository, data, validation
Every step is inspectable

Next up: Capstone: The Pipeline — prepare, validate, train, track

❮ Team Practices Capstone: The Pipeline ❯