MLOps Implementation Guide¶

from feast import Feature, FeatureView, FileSource

# Define feature source
customer_source = FileSource(
    path="data/customer_features.parquet",
    timestamp_field="event_timestamp"
)

# Define feature view
customer_features = FeatureView(
    name="customer_features",
    entities=["customer_id"],
    features=[
        Feature(name="total_transactions", dtype=Float32),
        Feature(name="avg_transaction_value", dtype=Float32),
        Feature(name="days_since_last_transaction", dtype=Int32),
    ],
    online=True,
    batch_source=customer_source,
)

# train.py
import mlflow
import yaml
from sklearn.model_selection import train_test_split
from sklearn.ensemble import GradientBoostingClassifier
import pandas as pd
import joblib

def train_model(config_path: str, data_path: str, output_path: str):
    """Train model with full logging."""

    # Load config
    with open(config_path) as f:
        config = yaml.safe_load(f)

    # Start MLflow run
    with mlflow.start_run():
        # Log config
        mlflow.log_params(config['model_params'])

        # Load and prepare data
        df = pd.read_parquet(data_path)
        X = df.drop('target', axis=1)
        y = df['target']

        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )

        # Train model
        model = GradientBoostingClassifier(**config['model_params'])
        model.fit(X_train, y_train)

        # Evaluate
        train_score = model.score(X_train, y_train)
        test_score = model.score(X_test, y_test)

        # Log metrics
        mlflow.log_metric("train_accuracy", train_score)
        mlflow.log_metric("test_accuracy", test_score)

        # Save model
        joblib.dump(model, output_path)
        mlflow.sklearn.log_model(model, "model")

        # Log model metadata
        mlflow.log_artifact(config_path)

        print(f"Model trained. Train: {train_score:.4f}, Test: {test_score:.4f}")

if __name__ == "__main__":
    train_model(
        config_path="config/model_config.yaml",
        data_path="data/features.parquet",
        output_path="models/model.pkl"
    )

# Model registration with MLflow
import mlflow

# Register model
model_uri = f"runs:/{run_id}/model"
model_version = mlflow.register_model(model_uri, "my_model")

# Transition to staging
client = mlflow.tracking.MlflowClient()
client.transition_model_version_stage(
    name="my_model",
    version=model_version.version,
    stage="Staging"
)

# After validation, promote to production
client.transition_model_version_stage(
    name="my_model",
    version=model_version.version,
    stage="Production"
)

# tests/test_model.py
import pytest
import numpy as np
from model import load_model, predict

class TestModel:

    @pytest.fixture
    def model(self):
        return load_model("models/model.pkl")

    @pytest.fixture
    def test_data(self):
        return np.random.rand(100, 47)  # 100 samples, 47 features

    def test_model_loads(self, model):
        """Model should load without error."""
        assert model is not None

    def test_prediction_shape(self, model, test_data):
        """Predictions should have correct shape."""
        predictions = predict(model, test_data)
        assert predictions.shape == (100,)

    def test_prediction_range(self, model, test_data):
        """Predictions should be valid probabilities."""
        predictions = predict(model, test_data)
        assert all(0 <= p <= 1 for p in predictions)

    def test_model_accuracy(self, model):
        """Model should meet minimum accuracy threshold."""
        # Load test data with known labels
        X_test, y_test = load_test_data()
        accuracy = model.score(X_test, y_test)
        assert accuracy >= 0.80, f"Accuracy {accuracy} below threshold"

    def test_fairness_metric(self, model):
        """Model should meet fairness threshold."""
        disparity = calculate_demographic_parity(model)
        assert disparity >= 0.80, f"Fairness disparity {disparity} below threshold"

# data_validation.py
from great_expectations import expect

def validate_training_data(df):
    """Validate training data quality."""

    expectations = []

    # Completeness checks
    expectations.append(
        expect(df['customer_id'].notnull().all(),
               "customer_id should not have nulls")
    )

    # Range checks
    expectations.append(
        expect(df['age'].between(0, 120).all(),
               "age should be between 0 and 120")
    )

    # Distribution checks
    expectations.append(
        expect(df['target'].mean() > 0.05,
               "target should have >5% positive class")
    )

    # Uniqueness checks
    expectations.append(
        expect(df['customer_id'].is_unique,
               "customer_id should be unique")
    )

    # Report results
    failed = [e for e in expectations if not e.success]
    if failed:
        raise DataValidationError(f"Data validation failed: {failed}")

    return True

# canary_deployment.py

class CanaryDeployer:
    """Gradual traffic shifting for model deployment."""

    def __init__(self, model_name: str):
        self.model_name = model_name
        self.canary_percentage = 0

    def start_canary(self, new_version: str, initial_percent: int = 5):
        """Start canary with small traffic percentage."""
        self.new_version = new_version
        self.canary_percentage = initial_percent
        self._update_routing()
        print(f"Canary started: {initial_percent}% to {new_version}")

    def increase_traffic(self, new_percent: int):
        """Increase canary traffic if metrics are good."""
        if self._check_metrics():
            self.canary_percentage = new_percent
            self._update_routing()
            print(f"Canary traffic increased to {new_percent}%")
        else:
            self.rollback()

    def promote(self):
        """Promote canary to full production."""
        if self._check_metrics():
            self.canary_percentage = 100
            self._update_routing()
            print(f"Canary promoted: {self.new_version} now at 100%")
        else:
            self.rollback()

    def rollback(self):
        """Rollback canary deployment."""
        self.canary_percentage = 0
        self._update_routing()
        print(f"Canary rolled back")

    def _check_metrics(self) -> bool:
        """Check if canary metrics are acceptable."""
        # Compare canary vs production metrics
        canary_metrics = get_canary_metrics(self.new_version)
        prod_metrics = get_production_metrics()

        # Check for degradation
        if canary_metrics['latency_p99'] > prod_metrics['latency_p99'] * 1.2:
            return False
        if canary_metrics['error_rate'] > prod_metrics['error_rate'] * 1.5:
            return False
        return True

# monitoring.py
from prometheus_client import Counter, Histogram, Gauge

# Request metrics
PREDICTION_COUNT = Counter(
    'ml_predictions_total',
    'Total predictions made',
    ['model_name', 'model_version', 'outcome']
)

PREDICTION_LATENCY = Histogram(
    'ml_prediction_latency_seconds',
    'Prediction latency',
    ['model_name'],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

# Model quality metrics
MODEL_ACCURACY = Gauge(
    'ml_model_accuracy',
    'Current model accuracy',
    ['model_name', 'model_version']
)

FEATURE_DRIFT = Gauge(
    'ml_feature_drift',
    'Feature drift score',
    ['model_name', 'feature_name']
)

PREDICTION_DRIFT = Gauge(
    'ml_prediction_drift',
    'Prediction distribution drift',
    ['model_name']
)