🚀 Introduction to MLOps: Operationalizing Machine Learning

📚 What You'll Learn

By the end of this lesson, you will be able to:

Explain what MLOps is and how it extends DevOps to the machine learning lifecycle
Map the ML lifecycle stages and the MLOps components that support each one
Track experiments, parameters, and metrics with MLflow
Version and register models so you can reproduce and promote them
Set up monitoring to detect data and model drift over time
Describe a CI/CD pipeline tailored to machine learning

⏱️ Estimated Time: 45–60 minutes

🎯 Project: Instrument a training script with MLflow to log parameters and metrics, register the resulting model, and outline the monitoring and CI/CD you'd add for production.

Introduction

MLOps (Machine Learning Operations) is a set of practices that combines Machine Learning, DevOps, and Data Engineering to deploy and maintain ML systems in production reliably and efficiently. It addresses the unique challenges of ML systems including data versioning, model versioning, experiment tracking, model monitoring, and automated retraining. This lesson introduces the core concepts, tools, and best practices for implementing MLOps.

The ML Lifecycle

flowchart LR A[Business Problem] --> B[Data Collection] B --> C[Data Preparation] C --> D[Feature Engineering] D --> E[Model Training] E --> F[Model Evaluation] F --> G{Good Enough?} G -->|No| H[Hyperparameter Tuning] H --> E G -->|Yes| I[Model Deployment] I --> J[Monitoring] J --> K[Model Performance] K --> L{Degradation?} L -->|Yes| M[Retrain] M --> E L -->|No| J style A fill:#e3f2fd style I fill:#c8e6c9 style J fill:#fff9c4 style M fill:#ffccbc

MLOps Components

graph TB A[MLOps Platform] --> B[Development] A --> C[Deployment] A --> D[Monitoring] B --> B1[Experiment Tracking] B --> B2[Version Control] B --> B3[Feature Store] B --> B4[Model Registry] C --> C1[CI/CD Pipelines] C --> C2[Model Serving] C --> C3[A/B Testing] C --> C4[Infrastructure] D --> D1[Data Drift] D --> D2[Model Performance] D --> D3[System Health] D --> D4[Alerting] style A fill:#f0f4c3 style B fill:#e1f5fe style C fill:#f3e5f5 style D fill:#fff3e0

Setting Up MLOps Environment

# Core MLOps libraries
import mlflow
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score
import joblib
import json
import os
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

print("MLOps Introduction")
print("=" * 50)
print(f"MLflow version: {mlflow.__version__}")

# Create project structure
def create_project_structure():
    """Create standard MLOps project structure"""
    directories = [
        'data/raw',
        'data/processed',
        'models',
        'notebooks',
        'src',
        'tests',
        'configs',
        'logs',
        'artifacts'
    ]
    
    for directory in directories:
        os.makedirs(directory, exist_ok=True)
    
    print("Project structure created:")
    for directory in directories:
        print(f"  ├── {directory}/")

create_project_structure()

Experiment Tracking with MLflow

class ExperimentTracker:
    """
    Experiment tracking with MLflow
    """
    
    def __init__(self, experiment_name="ml_experiments"):
        self.experiment_name = experiment_name
        mlflow.set_experiment(experiment_name)
        
    def train_with_tracking(self, X_train, X_test, y_train, y_test, params):
        """
        Train model with experiment tracking
        """
        with mlflow.start_run():
            # Log parameters
            mlflow.log_params(params)
            
            # Train model
            model = RandomForestClassifier(**params)
            
            # Track training time
            start_time = datetime.now()
            model.fit(X_train, y_train)
            training_time = (datetime.now() - start_time).total_seconds()
            
            # Make predictions
            y_pred = model.predict(X_test)
            
            # Calculate metrics
            metrics = {
                'accuracy': accuracy_score(y_test, y_pred),
                'precision': precision_score(y_test, y_pred, average='weighted'),
                'recall': recall_score(y_test, y_pred, average='weighted'),
                'f1': f1_score(y_test, y_pred, average='weighted'),
                'training_time': training_time
            }
            
            # Log metrics
            mlflow.log_metrics(metrics)
            
            # Log model
            mlflow.sklearn.log_model(
                model, 
                "model",
                registered_model_name=f"RandomForest_{self.experiment_name}"
            )
            
            # Log additional artifacts
            feature_importance = pd.DataFrame({
                'feature': X_train.columns,
                'importance': model.feature_importances_
            }).sort_values('importance', ascending=False)
            
            # Save and log feature importance
            feature_importance.to_csv('artifacts/feature_importance.csv', index=False)
            mlflow.log_artifact('artifacts/feature_importance.csv')
            
            print(f"Run ID: {mlflow.active_run().info.run_id}")
            print(f"Metrics: {metrics}")
            
            return model, metrics
    
    def run_experiments(self, X, y):
        """
        Run multiple experiments with different parameters
        """
        # Split data
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )
        
        # Define parameter grid
        param_grid = [
            {'n_estimators': 50, 'max_depth': 5, 'random_state': 42},
            {'n_estimators': 100, 'max_depth': 10, 'random_state': 42},
            {'n_estimators': 200, 'max_depth': 15, 'random_state': 42},
            {'n_estimators': 100, 'max_depth': None, 'random_state': 42}
        ]
        
        results = []
        
        print("\nRunning experiments...")
        print("-" * 40)
        
        for i, params in enumerate(param_grid, 1):
            print(f"\nExperiment {i}: {params}")
            model, metrics = self.train_with_tracking(
                X_train, X_test, y_train, y_test, params
            )
            results.append({
                'experiment': i,
                'params': params,
                'metrics': metrics
            })
        
        return results
    
    def compare_experiments(self, results):
        """
        Compare experiment results
        """
        import matplotlib.pyplot as plt
        
        # Create comparison DataFrame
        comparison = []
        for result in results:
            row = {'experiment': result['experiment']}
            row.update(result['params'])
            row.update(result['metrics'])
            comparison.append(row)
        
        df = pd.DataFrame(comparison)
        
        # Visualization
        fig, axes = plt.subplots(2, 2, figsize=(12, 10))
        
        # Accuracy comparison
        axes[0, 0].bar(df['experiment'], df['accuracy'])
        axes[0, 0].set_xlabel('Experiment')
        axes[0, 0].set_ylabel('Accuracy')
        axes[0, 0].set_title('Model Accuracy Comparison')
        axes[0, 0].grid(True, alpha=0.3)
        
        # F1 score comparison
        axes[0, 1].bar(df['experiment'], df['f1'], color='green')
        axes[0, 1].set_xlabel('Experiment')
        axes[0, 1].set_ylabel('F1 Score')
        axes[0, 1].set_title('F1 Score Comparison')
        axes[0, 1].grid(True, alpha=0.3)
        
        # Training time vs n_estimators
        axes[1, 0].scatter(df['n_estimators'], df['training_time'], s=100)
        axes[1, 0].set_xlabel('Number of Estimators')
        axes[1, 0].set_ylabel('Training Time (s)')
        axes[1, 0].set_title('Training Time vs Model Complexity')
        axes[1, 0].grid(True, alpha=0.3)
        
        # Accuracy vs max_depth
        max_depth_values = df['max_depth'].fillna(20)  # Replace None with 20 for viz
        axes[1, 1].scatter(max_depth_values, df['accuracy'], s=100, color='orange')
        axes[1, 1].set_xlabel('Max Depth')
        axes[1, 1].set_ylabel('Accuracy')
        axes[1, 1].set_title('Accuracy vs Tree Depth')
        axes[1, 1].grid(True, alpha=0.3)
        
        plt.suptitle('Experiment Results Comparison', fontsize=14)
        plt.tight_layout()
        plt.show()
        
        return df

# Demonstrate experiment tracking
print("\n" + "=" * 50)
print("EXPERIMENT TRACKING DEMO")
print("=" * 50)

# Generate sample data
from sklearn.datasets import make_classification

X, y = make_classification(
    n_samples=1000, n_features=20, n_informative=15,
    n_redundant=5, random_state=42
)

# Convert to DataFrame for better tracking
feature_names = [f'feature_{i}' for i in range(20)]
X_df = pd.DataFrame(X, columns=feature_names)

# Initialize tracker and run experiments
tracker = ExperimentTracker("intro_mlops")
results = tracker.run_experiments(X_df, y)

# Compare results
print("\n" + "=" * 50)
print("EXPERIMENT COMPARISON")
print("=" * 50)
comparison_df = tracker.compare_experiments(results)
print("\nResults Summary:")
print(comparison_df[['experiment', 'n_estimators', 'max_depth', 'accuracy', 'f1']])

Model Versioning and Registry

class ModelRegistry:
    """
    Model versioning and registry management
    """
    
    def __init__(self, registry_path="models"):
        self.registry_path = registry_path
        os.makedirs(registry_path, exist_ok=True)
        self.metadata_file = os.path.join(registry_path, "registry.json")
        self.load_registry()
    
    def load_registry(self):
        """Load or create model registry"""
        if os.path.exists(self.metadata_file):
            with open(self.metadata_file, 'r') as f:
                self.registry = json.load(f)
        else:
            self.registry = {"models": {}}
    
    def save_registry(self):
        """Save registry metadata"""
        with open(self.metadata_file, 'w') as f:
            json.dump(self.registry, f, indent=2, default=str)
    
    def register_model(self, model, model_name, version, metrics, metadata=None):
        """
        Register a new model version
        """
        # Create version directory
        version_path = os.path.join(self.registry_path, model_name, f"v{version}")
        os.makedirs(version_path, exist_ok=True)
        
        # Save model
        model_file = os.path.join(version_path, "model.pkl")
        joblib.dump(model, model_file)
        
        # Create model metadata
        model_info = {
            "name": model_name,
            "version": version,
            "path": model_file,
            "timestamp": datetime.now().isoformat(),
            "metrics": metrics,
            "metadata": metadata or {},
            "model_type": type(model).__name__
        }
        
        # Update registry
        if model_name not in self.registry["models"]:
            self.registry["models"][model_name] = {}
        
        self.registry["models"][model_name][f"v{version}"] = model_info
        
        # Mark as latest version
        self.registry["models"][model_name]["latest"] = f"v{version}"
        
        # Save registry
        self.save_registry()
        
        print(f"Model registered: {model_name} v{version}")
        print(f"Location: {model_file}")
        
        return model_info
    
    def load_model(self, model_name, version="latest"):
        """
        Load a specific model version
        """
        if model_name not in self.registry["models"]:
            raise ValueError(f"Model {model_name} not found in registry")
        
        if version == "latest":
            version = self.registry["models"][model_name]["latest"]
        
        model_info = self.registry["models"][model_name][version]
        model = joblib.load(model_info["path"])
        
        print(f"Loaded model: {model_name} {version}")
        print(f"Metrics: {model_info['metrics']}")
        
        return model, model_info
    
    def list_models(self):
        """List all registered models"""
        print("\n" + "=" * 50)
        print("MODEL REGISTRY")
        print("=" * 50)
        
        for model_name, versions in self.registry["models"].items():
            print(f"\n{model_name}:")
            latest = versions.get("latest", "None")
            
            for version_key, version_info in versions.items():
                if version_key != "latest" and isinstance(version_info, dict):
                    is_latest = " (latest)" if version_key == latest else ""
                    metrics = version_info.get("metrics", {})
                    print(f"  {version_key}{is_latest}: accuracy={metrics.get('accuracy', 'N/A'):.3f}")
    
    def compare_versions(self, model_name):
        """
        Compare different versions of a model
        """
        if model_name not in self.registry["models"]:
            print(f"Model {model_name} not found")
            return
        
        versions_data = []
        for version_key, version_info in self.registry["models"][model_name].items():
            if version_key != "latest" and isinstance(version_info, dict):
                versions_data.append({
                    'version': version_key,
                    'timestamp': version_info['timestamp'],
                    'accuracy': version_info['metrics'].get('accuracy', 0),
                    'f1': version_info['metrics'].get('f1', 0)
                })
        
        if versions_data:
            df = pd.DataFrame(versions_data)
            print(f"\nVersion comparison for {model_name}:")
            print(df.to_string(index=False))

# Demonstrate model registry
print("\n" + "=" * 50)
print("MODEL REGISTRY DEMO")
print("=" * 50)

registry = ModelRegistry()

# Register models from our experiments
for i, result in enumerate(results[:3], 1):  # Register first 3 models
    # Train a simple model for demo
    model = RandomForestClassifier(**result['params'])
    X_train, X_test, y_train, y_test = train_test_split(X_df, y, test_size=0.2, random_state=42)
    model.fit(X_train, y_train)
    
    registry.register_model(
        model=model,
        model_name="random_forest_classifier",
        version=i,
        metrics=result['metrics'],
        metadata={"experiment_id": result['experiment']}
    )

# List all models
registry.list_models()

# Load latest model
print("\n" + "=" * 50)
print("LOADING MODEL FROM REGISTRY")
print("=" * 50)
loaded_model, loaded_info = registry.load_model("random_forest_classifier", version="latest")

Data and Model Monitoring

flowchart TB A[Production Model] --> B[Monitor] B --> C[Data Quality] C --> C1[Missing Values] C --> C2[Data Types] C --> C3[Value Ranges] B --> D[Data Drift] D --> D1[Feature Distribution] D --> D2[Statistical Tests] D --> D3[Drift Score] B --> E[Model Performance] E --> E1[Accuracy Metrics] E --> E2[Business Metrics] E --> E3[Latency] B --> F[System Health] F --> F1[CPU/Memory] F --> F2[Request Rate] F --> F3[Error Rate] C --> G{Alert?} D --> G E --> G F --> G G -->|Yes| H[Trigger Action] H --> I[Retrain/Update/Scale] style A fill:#e3f2fd style G fill:#ffccbc style I fill:#c8e6c9

class ModelMonitor:
    """
    Monitor model performance and data drift
    """
    
    def __init__(self, reference_data, reference_predictions):
        self.reference_data = reference_data
        self.reference_predictions = reference_predictions
        self.monitoring_results = []
        
    def check_data_quality(self, new_data):
        """
        Check data quality issues
        """
        issues = []
        
        # Check for missing values
        missing = new_data.isnull().sum()
        if missing.any():
            issues.append(f"Missing values found: {missing[missing > 0].to_dict()}")
        
        # Check data types
        ref_dtypes = self.reference_data.dtypes
        new_dtypes = new_data.dtypes
        if not ref_dtypes.equals(new_dtypes):
            issues.append("Data type mismatch detected")
        
        # Check value ranges
        for col in new_data.columns:
            if col in self.reference_data.columns:
                ref_min, ref_max = self.reference_data[col].min(), self.reference_data[col].max()
                new_min, new_max = new_data[col].min(), new_data[col].max()
                
                if new_min < ref_min * 0.9 or new_max > ref_max * 1.1:
                    issues.append(f"Value range drift in {col}: [{new_min:.2f}, {new_max:.2f}]")
        
        return issues
    
    def calculate_drift(self, new_data):
        """
        Calculate feature drift using statistical tests
        """
        from scipy import stats
        
        drift_scores = {}
        
        for col in new_data.columns:
            if col in self.reference_data.columns:
                # Kolmogorov-Smirnov test
                ks_statistic, p_value = stats.ks_2samp(
                    self.reference_data[col],
                    new_data[col]
                )
                
                drift_scores[col] = {
                    'ks_statistic': ks_statistic,
                    'p_value': p_value,
                    'is_drifted': p_value < 0.05
                }
        
        return drift_scores
    
    def monitor_performance(self, model, new_data, new_labels):
        """
        Monitor model performance on new data
        """
        # Make predictions
        predictions = model.predict(new_data)
        
        # Calculate metrics
        performance = {
            'accuracy': accuracy_score(new_labels, predictions),
            'precision': precision_score(new_labels, predictions, average='weighted'),
            'recall': recall_score(new_labels, predictions, average='weighted'),
            'f1': f1_score(new_labels, predictions, average='weighted')
        }
        
        # Compare with baseline
        baseline_accuracy = accuracy_score(
            self.reference_predictions[:len(new_labels)], 
            self.reference_predictions[:len(new_labels)]
        )
        
        performance['accuracy_drop'] = baseline_accuracy - performance['accuracy']
        
        return performance
    
    def generate_monitoring_report(self, model, new_data, new_labels):
        """
        Generate comprehensive monitoring report
        """
        print("\n" + "=" * 50)
        print("MODEL MONITORING REPORT")
        print("=" * 50)
        print(f"Timestamp: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        
        # Data quality check
        print("\n1. DATA QUALITY CHECK")
        print("-" * 30)
        quality_issues = self.check_data_quality(new_data)
        if quality_issues:
            for issue in quality_issues:
                print(f"  ⚠️ {issue}")
        else:
            print("  ✅ No data quality issues detected")
        
        # Data drift check
        print("\n2. DATA DRIFT ANALYSIS")
        print("-" * 30)
        drift_scores = self.calculate_drift(new_data)
        drifted_features = [f for f, scores in drift_scores.items() if scores['is_drifted']]
        
        if drifted_features:
            print(f"  ⚠️ Drift detected in {len(drifted_features)} features:")
            for feature in drifted_features[:5]:  # Show top 5
                print(f"    - {feature}: KS={drift_scores[feature]['ks_statistic']:.3f}")
        else:
            print("  ✅ No significant drift detected")
        
        # Performance monitoring
        print("\n3. MODEL PERFORMANCE")
        print("-" * 30)
        performance = self.monitor_performance(model, new_data, new_labels)
        
        print(f"  Accuracy: {performance['accuracy']:.3f}")
        print(f"  Precision: {performance['precision']:.3f}")
        print(f"  Recall: {performance['recall']:.3f}")
        print(f"  F1-Score: {performance['f1']:.3f}")
        
        if performance['accuracy_drop'] > 0.05:
            print(f"  ⚠️ Performance degradation detected: {performance['accuracy_drop']:.3f}")
        
        # Recommendations
        print("\n4. RECOMMENDATIONS")
        print("-" * 30)
        
        if drifted_features or performance['accuracy_drop'] > 0.05:
            print("  🔄 Consider retraining the model")
        if quality_issues:
            print("  🔧 Fix data quality issues before prediction")
        if not (drifted_features or quality_issues or performance['accuracy_drop'] > 0.05):
            print("  ✅ Model performing well, no action needed")
        
        return {
            'timestamp': datetime.now(),
            'quality_issues': quality_issues,
            'drift_scores': drift_scores,
            'performance': performance
        }

# Demonstrate monitoring
print("\n" + "=" * 50)
print("MODEL MONITORING DEMO")
print("=" * 50)

# Create reference data
X_ref, X_new, y_ref, y_new = train_test_split(X_df, y, test_size=0.3, random_state=42)

# Train reference model
ref_model = RandomForestClassifier(n_estimators=100, random_state=42)
ref_model.fit(X_ref, y_ref)
ref_predictions = ref_model.predict(X_ref)

# Initialize monitor
monitor = ModelMonitor(X_ref, ref_predictions)

# Simulate new data with some drift
X_new_drifted = X_new.copy()
X_new_drifted.iloc[:, :5] = X_new_drifted.iloc[:, :5] * 1.5  # Introduce drift

# Generate monitoring report
report = monitor.generate_monitoring_report(ref_model, X_new_drifted, y_new)

CI/CD Pipeline for ML

# Example CI/CD configuration (GitHub Actions)
cicd_yaml = """
name: ML Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test-and-train:
    runs-on: ubuntu-latest
    
    steps:
    - uses: actions/checkout@v2
    
    - name: Set up Python
      uses: actions/setup-python@v2
      with:
        python-version: '3.8'
    
    - name: Install dependencies
      run: |
        pip install -r requirements.txt
    
    - name: Run tests
      run: |
        pytest tests/
    
    - name: Data validation
      run: |
        python src/validate_data.py
    
    - name: Train model
      run: |
        python src/train_model.py
    
    - name: Evaluate model
      run: |
        python src/evaluate_model.py
    
    - name: Register model
      if: github.ref == 'refs/heads/main'
      run: |
        python src/register_model.py
    
    - name: Deploy model
      if: github.ref == 'refs/heads/main'
      run: |
        python src/deploy_model.py
"""

print("\n" + "=" * 50)
print("EXAMPLE CI/CD PIPELINE")
print("=" * 50)
print("GitHub Actions workflow for ML pipeline:")
print(cicd_yaml)

Best Practices

🎯 MLOps Best Practices

Version Everything: Code, data, models, and configurations
Automate Pipeline: From data ingestion to model deployment
Monitor Continuously: Track data quality, drift, and performance
Test Rigorously: Unit tests, integration tests, and model validation
Document Thoroughly: Experiments, decisions, and model behavior
Reproducible Results: Use seeds, log environments, containerize
Feature Store: Centralize feature engineering and serving
A/B Testing: Validate model improvements in production
Rollback Strategy: Quick reversion to previous model versions
Security: Encrypt models, secure APIs, audit access

Practice Exercises

Exercise 1: Build a Complete MLOps Pipeline

Create an end-to-end MLOps pipeline:

Set up experiment tracking with MLflow
Implement automated model training
Create model registry with versioning
Build monitoring dashboard
Deploy with API endpoint

Exercise 2: Implement Data Drift Detection

Build a drift detection system:

Collect reference data distribution
Implement statistical tests for drift
Create alerting mechanism
Automate retraining triggers

Summary

✅ You've Learned

What MLOps is and why it's essential
The complete ML lifecycle from development to production
Experiment tracking and model versioning
Model registry and management
Data quality and drift monitoring
Performance monitoring and alerting
CI/CD pipelines for ML
Best practices for production ML systems

📓 Learning Journal

Keep a learning journal — digital or physical. After this lesson, take a few minutes to write down:

Key concepts you learned
Techniques that clicked for you
Questions or confusion points to revisit
Ideas you want to try
Your progress and feelings about learning this

✍️ This lesson's prompt: Think back to a model or experiment you built without tracking anything. If you had to reproduce it exactly six months later, what would be missing — and how would MLOps practices have saved you?

📝 Lesson Summary

🎓 Key Takeaways

MLOps applies DevOps discipline — automation, versioning, testing, monitoring — to the full ML lifecycle, not just the code.
Experiment tracking (e.g. MLflow) records parameters, metrics, and artifacts so results are reproducible and comparable.
Model versioning and a registry let you promote, roll back, and audit which model is in production.
Monitoring for data and model drift, plus CI/CD, keeps deployed models reliable as the world changes.

🎉 What You've Accomplished

You now understand what it takes to run machine learning as a dependable, repeatable practice — tracking experiments, versioning models, and planning the monitoring and automation that keep production models healthy.

❓ Common Questions at This Stage

How is MLOps different from DevOps?

MLOps includes everything DevOps does, plus the parts unique to ML: versioning data and models (not just code), tracking experiments, and monitoring for drift where model performance decays even when the code doesn't change.

Do small projects really need MLOps?

Start light. Even a solo project benefits from experiment tracking and model versioning. Add heavier automation, registries, and monitoring as the project's stakes and lifespan grow.

What is model drift?

It's when a deployed model's performance degrades because incoming data (data drift) or the underlying relationships (concept drift) have changed since training. Monitoring catches it so you can retrain in time.

🔭 Looking Ahead

With the MLOps big picture in place, the next step gets concrete — packaging and deploying a model as a live service that real applications can call.

✅ Before the Next Lesson

Add MLflow logging to a training script and compare two runs in the tracking UI.
Register your best run's model and note how you'd promote it from staging to production.
Write your Learning Journal entry for this lesson

🌟 Encouragement for the Journey

MLOps is what turns clever experiments into systems people can rely on. Learning it is how you go from "I trained a model" to "I run machine learning that lasts." You're stepping up to the professional level — keep going!