MLOps Best Practices:Production ML at Scale

# .gitlab-ci.yml
stages:
  - data_validation
  - feature_engineering
  - model_training
  - model_evaluation
  - model_deployment
  - monitoring

data_validation:
  stage: data_validation
  script:
    - python scripts/validate_data.py
    - dvc pull
    - great_expectations checkpoint run data_quality
  artifacts:
    reports:
      - reports/data_validation.html

model_training:
  stage: model_training
  script:
    - python src/train.py --config configs/model_config.yaml
    - mlflow run . -P epochs=100 -P batch_size=32
  artifacts:
    paths:
      - models/
      - metrics/
  only:
    - main
    - develop

model_deployment:
  stage: model_deployment
  script:
    - docker build -t model:$CI_COMMIT_SHA .
    - kubectl apply -f k8s/deployment.yaml
    - kubectl set image deployment/model model=model:$CI_COMMIT_SHA
  environment:
    name: production
    url: https://api.example.com/model
  when: manual
  only:
    - main

# monitoring/model_monitor.py
import numpy as np
from evidently import ColumnMapping
from evidently.report import Report
from evidently.metrics import *

class ModelMonitor:
    def __init__(self, reference_data, model_name):
        self.reference = reference_data
        self.model_name = model_name
        self.alerts = []
        
    def check_data_drift(self, current_data):
        """Check for data drift between reference and current data"""
        report = Report(metrics=[
            DataDriftTable(),
            DataQualityTable(),
            RegressionQualityMetric()
        ])
        
        report.run(
            reference_data=self.reference,
            current_data=current_data,
            column_mapping=self.column_mapping
        )
        
        drift_score = report.get_metric("DataDriftTable").drift_share
        
        if drift_score > 0.3:
            self.trigger_alert(
                severity="HIGH",
                message=f"Data drift detected: {drift_score:.2%}"
            )
            
        return report
    
    def check_performance_degradation(self, predictions, actuals):
        """Monitor model performance metrics"""
        mae = np.mean(np.abs(predictions - actuals))
        rmse = np.sqrt(np.mean((predictions - actuals)**2))
        
        if mae > self.thresholds['mae']:
            self.trigger_alert(
                severity="MEDIUM",
                message=f"MAE threshold exceeded: {mae:.3f}"
            )
            
        return {"mae": mae, "rmse": rmse}
    
    def trigger_alert(self, severity, message):
        """Send alerts to monitoring systems"""
        alert = {
            "model": self.model_name,
            "severity": severity,
            "message": message,
            "timestamp": datetime.now()
        }
        
        # Send to Prometheus AlertManager
        self.send_to_alertmanager(alert)
        
        # Log to centralized logging
        logger.error(f"Model Alert: {alert}")
        
        self.alerts.append(alert)

# feature_store/features.py
from feast import FeatureStore, Entity, Feature, FeatureView
from feast.types import Float32, Int64, String
import pandas as pd

# Initialize feature store
store = FeatureStore(repo_path="feature_repo/")

# Define customer entity
customer = Entity(
    name="customer",
    value_type=Int64,
    description="Customer ID"
)

# Define feature views
customer_features = FeatureView(
    name="customer_features",
    entities=["customer"],
    ttl=timedelta(days=1),
    features=[
        Feature(name="total_purchases", dtype=Float32),
        Feature(name="days_since_last_purchase", dtype=Int64),
        Feature(name="customer_segment", dtype=String),
        Feature(name="lifetime_value", dtype=Float32),
    ],
    online=True,
    batch_source=BigQuerySource(
        table="project.dataset.customer_features",
        timestamp_column="event_timestamp"
    ),
    stream_source=KafkaSource(
        topic="customer_events",
        format="avro"
    )
)

# Training data retrieval
def get_training_data(entity_df, feature_refs):
    """Retrieve historical features for training"""
    training_data = store.get_historical_features(
        entity_df=entity_df,
        feature_refs=feature_refs
    ).to_df()
    
    return training_data

# Online serving
def get_online_features(customer_ids):
    """Retrieve features for real-time inference"""
    feature_vector = store.get_online_features(
        feature_refs=[
            "customer_features:total_purchases",
            "customer_features:days_since_last_purchase",
            "customer_features:customer_segment",
            "customer_features:lifetime_value"
        ],
        entity_rows=[{"customer": id} for id in customer_ids]
    )
    
    return feature_vector.to_dict()