majiayu000 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:
# Rollback
delete_deployment(f"{service_name}-canary")
return False
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'] > 0.9):
deploy_model(model_path=train_task.outputs['model_path'])
# Compile and run
kfp.compiler.Compiler().compile(ml_pipeline, 'pipeline.yaml')
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 hasattr(model, 'explain') or hasattr(model, 'feature_importances_')
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
- Foundations: CI/CD basics, containerization, version control
- ML Pipelines: Experiment tracking, model registry, orchestration
- Deployment: Kubernetes, serving infrastructure, scaling
- Monitoring: Drift detection, performance tracking, alerting
- Governance: Compliance, auditing, model documentation
See references/ for pipeline templates, monitoring dashboards, and deployment configurations.