mlops-advanced

name

mlops-advanced

description

Implement advanced MLOps practices for production ML systems. Use for: building CI/CD pipelines for ML models, implementing continuous training and monitoring, managing model registries and versioning, deploying with blue-green and canary strategies, monitoring model drift and performance degradation, orchestrating ML workflows with Kubeflow/Airflow, implementing feature stores, ensuring model governance and compliance, and scaling ML infrastructure with Kubernetes.

MLOps Advanced

Implement production-grade MLOps practices for scalable, reliable machine learning systems.

Overview

MLOps (Machine Learning Operations) extends DevOps principles to machine learning, automating and streamlining the entire ML lifecycle from development to deployment and monitoring. Advanced MLOps addresses unique ML challenges including data dependencies, model versioning, continuous training, performance monitoring, and infrastructure scaling. It encompasses CI/CD for ML, automated retraining, drift detection, and governance frameworks.

MLOps Maturity Levels

Level 0 - Manual Process:

• Manual data preparation and training
• Notebook-driven development
• Manual deployment
• No CI/CD or monitoring

Level 1 - ML Pipeline Automation:

• Automated training pipelines
• Continuous training (CT)
• Model and data validation
• Automated deployment of prediction service

Level 2 - CI/CD Pipeline Automation:

• Full CI/CD system for ML
• Automated testing (data, model, infrastructure)
• Rapid experimentation and deployment
• Comprehensive monitoring and logging

CI/CD for Machine Learning

Continuous Integration Components

# .github/workflows/ml-ci.yml
name: ML CI Pipeline

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      
      - name: Data Validation
        run: |
          python scripts/validate_data.py
          python scripts/check_data_drift.py
      
      - name: Feature Validation
        run: python scripts/validate_features.py
      
      - name: Model Training
        run: python scripts/train_model.py
      
      - name: Model Evaluation
        run: |
          python scripts/evaluate_model.py
          python scripts/check_fairness.py
      
      - name: Unit Tests
        run: pytest tests/
      
      - name: Integration Tests
        run: pytest tests/integration/

Continuous Deployment

# deploy_model.py
import mlflow
from kubernetes import client, config

def deploy_model(model_uri, deployment_name, namespace='production'):
    # Load model from registry
    model = mlflow.pyfunc.load_model(model_uri)
    
    # Create deployment configuration
    deployment = create_k8s_deployment(
        name=deployment_name,
        image=f'ml-model-server:{model.metadata.run_id}',
        replicas=3,
        resources={'cpu': '2', 'memory': '4Gi'}
    )
    
    # Deploy to Kubernetes
    config.load_kube_config()
    apps_v1 = client.AppsV1Api()
    apps_v1.create_namespaced_deployment(namespace, deployment)
    
    # Create service
    service = create_k8s_service(deployment_name)
    core_v1 = client.CoreV1Api()
    core_v1.create_namespaced_service(namespace, service)

Model Registry and Versioning

MLflow Model Registry

import mlflow
from mlflow.tracking import MlflowClient

# Register model
mlflow.set_tracking_uri("http://mlflow-server:5000")
client = MlflowClient()

# Log and register model
with mlflow.start_run():
    mlflow.log_params(params)
    mlflow.log_metrics(metrics)
    mlflow.sklearn.log_model(model, "model")
    
    # Register to model registry
    model_uri = f"runs:/{mlflow.active_run().info.run_id}/model"
    mlflow.register_model(model_uri, "my_model")

# Transition model stage
client.transition_model_version_stage(
    name="my_model",
    version=1,
    stage="Production"
)

# Load production model
model = mlflow.pyfunc.load_model("models:/my_model/Production")

Model Versioning Strategy

class ModelVersion:
    def __init__(self, model_name, version, metadata):
        self.model_name = model_name
        self.version = version
        self.metadata = {
            'training_date': metadata['date'],
            'dataset_version': metadata['data_version'],
            'metrics': metadata['metrics'],
            'hyperparameters': metadata['params'],
            'framework': metadata['framework'],
            'dependencies': metadata['dependencies']
        }
    
    def compare_with(self, other_version):
        return {
            'accuracy_diff': self.metadata['metrics']['accuracy'] - 
                           other_version.metadata['metrics']['accuracy'],
            'latency_diff': self.metadata['metrics']['latency'] - 
                          other_version.metadata['metrics']['latency']
        }

Deployment Strategies

Blue-Green Deployment

def blue_green_deployment(new_model_uri, service_name):
    # Deploy green (new) version
    deploy_model(new_model_uri, f"{service_name}-green")
    
    # Run smoke tests
    if not run_smoke_tests(f"{service_name}-green"):
        rollback(f"{service_name}-green")
        return False
    
    # Switch traffic from blue to green
    update_service_selector(service_name, version='green')
    
    # Monitor for issues
    if monitor_deployment(service_name, duration=300):
        # Success - remove blue deployment
        delete_deployment(f"{service_name}-blue")
        return True
    else:
        # Rollback to blue
        update_service_selector(service_name, version='blue')
        delete_deployment(f"{service_name}-green")
        return False

Canary Deployment

def canary_deployment(new_model_uri, service_name, canary_percentage=10):
    # Deploy canary version
    deploy_model(new_model_uri, f"{service_name}-canary", replicas=1)
    
    # Route small percentage of traffic to canary
    update_traffic_split(service_name, {
        'stable': 100 - canary_percentage,
        'canary': canary_percentage
    })
    
    # Monitor canary metrics
    canary_metrics = monitor_canary(f"{service_name}-canary", duration=600)
    stable_metrics = monitor_canary(f"{service_name}-stable", duration=600)
    
    if canary_metrics['error_rate'] <= stable_metrics['error_rate'] * 1.1:
        # Gradually increase canary traffic
        for percentage in [25, 50, 75, 100]:
            update_traffic_split(service_name, {
                'stable': 100 - percentage,
                'canary': percentage
            })
            time.sleep(300)
        
        # Replace stable with canary
        delete_deployment(f"{service_name}-stable")
        rename_deployment(f"{service_name}-canary", f"{service_name}-stable")
        return True
    else:
        
        delete_deployment()
         

Continuous Monitoring

Model Performance Monitoring

import prometheus_client as prom

# Define metrics
prediction_latency = prom.Histogram('model_prediction_latency_seconds', 
                                    'Model prediction latency')
prediction_counter = prom.Counter('model_predictions_total', 
                                 'Total predictions', ['model_version'])
error_counter = prom.Counter('model_errors_total', 
                            'Total errors', ['error_type'])

class MonitoredModel:
    def __init__(self, model, version):
        self.model = model
        self.version = version
    
    @prediction_latency.time()
    def predict(self, features):
        try:
            prediction = self.model.predict(features)
            prediction_counter.labels(model_version=self.version).inc()
            return prediction
        except Exception as e:
            error_counter.labels(error_type=type(e).__name__).inc()
            raise

Data Drift Detection

from scipy.stats import ks_2samp
import numpy as np

class DriftDetector:
    def __init__(self, reference_data, threshold=0.05):
        self.reference_data = reference_data
        self.threshold = threshold
    
    def detect_drift(self, current_data):
        drift_detected = {}
        
        for feature in self.reference_data.columns:
            # Kolmogorov-Smirnov test
            statistic, p_value = ks_2samp(
                self.reference_data[feature],
                current_data[feature]
            )
            
            drift_detected[feature] = {
                'drift': p_value < self.threshold,
                'p_value': p_value,
                'statistic': statistic
            }
        
        return drift_detected
    
    def calculate_psi(self, reference, current, bins=10):
        # Population Stability Index
        ref_hist, bin_edges = np.histogram(reference, bins=bins)
        cur_hist, _ = np.histogram(current, bins=bin_edges)
        
        ref_pct = ref_hist / len(reference)
        cur_pct = cur_hist / len(current)
        
        psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / (ref_pct + 1e-10)))
        return psi

Model Drift Detection

class ModelDriftDetector:
    def __init__(self, baseline_metrics):
        self.baseline_metrics = baseline_metrics
    
    def detect_performance_drift(self, current_metrics):
        drift_alerts = []
        
        for metric, baseline_value in self.baseline_metrics.items():
            current_value = current_metrics[metric]
            drift_percentage = abs(current_value - baseline_value) / baseline_value * 100
            
            if drift_percentage > 10:  # 10% threshold
                drift_alerts.append({
                    'metric': metric,
                    'baseline': baseline_value,
                    'current': current_value,
                    'drift_percentage': drift_percentage
                })
        
        return drift_alerts

Feature Store

from feast import FeatureStore, Entity, FeatureView, Field
from feast.types import Float32, Int64
from datetime import timedelta

# Define entity
user = Entity(name="user", join_keys=["user_id"])

# Define feature view
user_features = FeatureView(
    name="user_features",
    entities=[user],
    ttl=timedelta(days=1),
    schema=[
        Field(name="age", dtype=Int64),
        Field(name="avg_purchase", dtype=Float32),
        Field(name="total_purchases", dtype=Int64)
    ],
    source=parquet_source
)

# Initialize feature store
store = FeatureStore(repo_path=".")

# Get online features
features = store.get_online_features(
    features=["user_features:age", "user_features:avg_purchase"],
    entity_rows=[{"user_id": 123}, {"user_id": 456}]
).to_dict()

# Get historical features for training
training_data = store.get_historical_features(
    entity_df=entity_df,
    features=["user_features:age", "user_features:avg_purchase"]
).to_df()

ML Pipeline Orchestration

Kubeflow Pipeline

import kfp
from kfp import dsl

@dsl.component
def preprocess_data(input_path: str, output_path: str):
    # Data preprocessing logic
    pass

@dsl.component
def train_model(data_path: str, model_path: str, params: dict):
    # Model training logic
    pass

@dsl.component
def evaluate_model(model_path: str, test_data_path: str) -> dict:
    # Model evaluation logic
    pass

@dsl.pipeline(name='ML Training Pipeline')
def ml_pipeline(input_data: str, model_output: str):
    preprocess_task = preprocess_data(input_path=input_data, output_path='/data/processed')
    
    train_task = train_model(
        data_path=preprocess_task.outputs['output_path'],
        model_path=model_output,
        params={'learning_rate': 0.001, 'epochs': 10}
    )
    
    evaluate_task = evaluate_model(
        model_path=train_task.outputs['model_path'],
        test_data_path='/data/test'
    )
    
    with dsl.Condition(evaluate_task.outputs['accuracy'] > ):
        deploy_model(model_path=train_task.outputs[])


kfp.compiler.Compiler().(ml_pipeline, )

Model Governance

class ModelGovernance:
    def __init__(self):
        self.approval_required = True
        self.compliance_checks = []
    
    def register_compliance_check(self, check_func):
        self.compliance_checks.append(check_func)
    
    def validate_model(self, model, metadata):
        results = {
            'fairness': self.check_fairness(model, metadata),
            'explainability': self.check_explainability(model),
            'performance': self.check_performance(metadata),
            'compliance': all(check(model, metadata) for check in self.compliance_checks)
        }
        
        return all(results.values()), results
    
    def check_fairness(self, model, metadata):
        # Check for bias across protected attributes
        fairness_metrics = metadata.get('fairness_metrics', {})
        return all(metric < 0.1 for metric in fairness_metrics.values())
    
    def check_explainability(self, model):
        # Ensure model has explanation capabilities
        return (model, )  (model, )

Best Practices

Pipeline Design:

• Modularize components for reusability
• Implement idempotent operations
• Use containerization for reproducibility
• Version all artifacts (data, code, models)

Monitoring:

• Track both technical and business metrics
• Set up automated alerts for drift
• Monitor model latency and throughput
• Log predictions for debugging

Deployment:

• Use gradual rollout strategies
• Implement automated rollback
• Maintain model lineage
• Document deployment procedures

Governance:

• Establish approval workflows
• Implement audit logging
• Ensure regulatory compliance
• Document model decisions

Tools and Platforms

Experiment Tracking: MLflow, Weights & Biases, Neptune.ai Model Registry: MLflow, Vertex AI, SageMaker Pipeline Orchestration: Kubeflow, Airflow, Prefect, Metaflow Feature Stores: Feast, Tecton, Hopsworks Monitoring: Prometheus, Grafana, Evidently, WhyLabs Deployment: Kubernetes, Docker, TensorFlow Serving, TorchServe Cloud Platforms: AWS SageMaker, Google Vertex AI, Azure ML

Learning Path

1. Foundations: CI/CD basics, containerization, version control
2. ML Pipelines: Experiment tracking, model registry, orchestration
3. Deployment: Kubernetes, serving infrastructure, scaling
4. Monitoring: Drift detection, performance tracking, alerting
5. Governance: Compliance, auditing, model documentation

See references/ for pipeline templates, monitoring dashboards, and deployment configurations.