MLOps: Machine Learning Operations Guide

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
import numpy as np

# Generate data
X_train = np.random.rand(100, 5)
y_train = np.random.randint(0, 2, 100)
X_test = np.random.rand(20, 5)
y_test = np.random.randint(0, 2, 20)

mlflow.set_experiment("classification-demo")

with mlflow.start_run():
    # Log parameters
    n_estimators = 100
    max_depth = 10
    mlflow.log_param("n_estimators", n_estimators)
    mlflow.log_param("max_depth", max_depth)

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

    # Log metrics
    preds = model.predict(X_test)
    acc = accuracy_score(y_test, preds)
    mlflow.log_metric("accuracy", acc)

    # Log model
    mlflow.sklearn.log_model(model, "model")

    print(f"Run ID: {mlflow.active_run().info.run_id}")
    print(f"Accuracy: {acc:.3f}")

import wandb

wandb.init(project="classification-demo", config={
    "n_estimators": 100,
    "max_depth": 10,
    "learning_rate": 0.01
})

# Log metrics during training
for epoch in range(10):
    loss = 1.0 / (epoch + 1)
    wandb.log({"epoch": epoch, "loss": loss, "accuracy": 0.5 + epoch * 0.05})

wandb.finish()
print("Run logged to Weights & Biases")

# Conceptual example using Feast (open-source feature store)
from datetime import datetime
import pandas as pd

# Define a feature view
feature_data = pd.DataFrame({
    "user_id": [1, 2, 3],
    "avg_session_duration": [120.5, 45.2, 300.1],
    "num_logins_7d": [15, 3, 42],
    "event_timestamp": [datetime.now()] * 3
})

# In production, features are served via Feast API
def get_online_features(user_id):
    # Feast retrieves pre-computed features in real-time
    return {
        "avg_session_duration": 120.5,
        "num_logins_7d": 15
    }

features = get_online_features(1)
print(f"Online features for user 1: {features}")

import mlflow

# Register a model version
client = mlflow.tracking.MlflowClient()

model_uri = "runs:/a1b2c3d4e5f6g7h8/model"
model_name = "classification-model"

result = mlflow.register_model(model_uri, model_name)
print(f"Registered model: {result.name} version {result.version}")

# List all versions
versions = client.get_latest_versions(model_name)
for v in versions:
    print(f"  Version {v.version}: stage={v.current_stage}, run_id={v.run_id}")

# .github/workflows/ml-pipeline.yml (conceptual)
name: ML Pipeline
on: [push]

jobs:
  train:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - run: pip install -r requirements.txt
      - run: python train.py  # Trains and logs to MLflow
      - run: python evaluate.py  # Evaluates, fails if metrics below threshold
      - run: python deploy.py  # Promotes to staging/production

from great_expectations.dataset import PandasDataset
import pandas as pd

# Validate incoming data
df = pd.DataFrame({
    "age": [25, -5, 30, 200],
    "income": [50000, 60000, None, 70000]
})

ds = PandasDataset(df)

# Define expectations
expectations = {
    "age > 0": ds.expect_column_values_to_be_between("age", 0, 120),
    "income not null": ds.expect_column_values_to_not_be_null("income"),
}

for check, result in expectations.items():
    status = "PASS" if result["success"] else "FAIL"
    print(f"{check}: {status} (expected {result['expectation_config']['expectation_type']})")

import time
import random

class ModelMonitor:
    def __init__(self, threshold=0.7):
        self.threshold = threshold
        self.predictions = []

    def log_prediction(self, features, pred, actual=None):
        self.predictions.append({
            "timestamp": time.time(),
            "features": features,
            "prediction": pred,
            "actual": actual
        })

    def calculate_accuracy(self):
        recent = [p for p in self.predictions if p["actual"] is not None]
        if not recent:
            return None
        correct = sum(1 for p in recent if p["prediction"] == p["actual"])
        return correct / len(recent)

    def alert_if_needed(self):
        acc = self.calculate_accuracy()
        if acc is not None and acc < self.threshold:
            print(f"ALERT: Accuracy dropped to {acc:.3f} (threshold: {self.threshold})")

monitor = ModelMonitor()
for i in range(100):
    monitor.log_prediction(
        features=[random.random() for _ in range(5)],
        pred=random.randint(0, 1),
        actual=random.randint(0, 1)
    )
monitor.alert_if_needed()
print(f"Current accuracy: {monitor.calculate_accuracy():.3f}")