ML Notes

CI/CD for Machine Learning

Implementing continuous integration and continuous deployment pipelines for ML models including testing, validation, and automated deployment.

CI/CD (Continuous Integration/Continuous Deployment) ensures reliable, reproducible ML pipelines and deployments. Unlike traditional software, ML systems have unique challenges: data drift, model degradation, and versioning complexity.

Why CI/CD for ML Matters

Traditional software CI/CD isn't enough for ML:

Code may not change but model performance can degrade (data drift)
Need to version data, features, models, and code
Testing must include data validation and model performance
Deployment requires A/B testing and gradual rollout

Example

Traditional Software: ML System: Code → Build → Test → Deploy Data + Code → Pipeline → Train → Test → Deploy ↑ Monitor, Detect Drift, Retrain ↓

ML Pipeline Architecture

┌─────────────────────────────┐ │ Data Collection │ │ (Raw, unprocessed data) │ └────────────┬────────────────┘ │ ▼ ┌─────────────────────────────┐ │ Data Validation │ │ • Schema checking │ │ • Outlier detection │ │ • Data quality metrics │ │ • Missing value handling │ └────────────┬────────────────┘ │ ▼ ┌─────────────────────────────┐ │ Feature Engineering │ │ • Feature calculations │ │ • Feature validation │ │ • Feature importance │ └────────────┬────────────────┘ │ ▼ ┌─────────────────────────────┐ │ Model Training │ │ • Hyperparameter tuning │ │ • Cross-validation │ │ • Model artifact storage │ └────────────┬────────────────┘ │ ▼ ┌─────────────────────────────┐ │ Model Validation │ │ • Performance on test set │ │ • Fairness checks │ │ • Robustness tests │ │ • Regression detection │ └────────────┬────────────────┘ │ ▼ ┌─────────────────────────────┐ │ Deployment │ │ • Shadow deployment │ │ • A/B testing │ │ • Canary rollout │ │ • Full production rollout │ └────────────┬────────────────┘ │ ▼ ┌─────────────────────────────┐ │ Monitoring │ │ • Model performance │ │ • Data drift │ │ • Alert & retrain triggers │ │ • Feedback collection │ └─────────────────────────────┘

Version Control Best Practices

What to Version Control

Example

✓ Code (models, preprocessing, evaluation scripts) ✓ Configurations (hyperparameters, feature lists) ✓ Test data (small samples for CI tests) ✓ Requirements/dependencies (requirements.txt, setup.py) ✓ Documentation (README, architecture diagrams) ✗ Model artifacts (large binaries, use model registry) ✗ Training data (large, use DVC or similar) ✗ Credentials (use environment variables, secrets) ✗ Build outputs (use .gitignore)

Git Workflow for ML

python example

# 1. Create feature branch
git checkout -b feature/new-model-type

# 2. Experiment locally
# ... modify code, train models ...

# 3. Commit with clear messages
git commit -m "Implement gradient boosting model with feature engineering"

# 4. Push to trigger CI tests
git push origin feature/new-model-type

# 5. Create pull request
# - Runs automated tests
# - Checks performance vs baseline
# - Code review
# - Merge to main

# 6. CD automatically deploys after merge

Automated Testing for ML

Data Validation Tests

Example

import pandas as pd import great_expectations as ge def validate_input_data(df): """Validate data schema and quality""" # Column existence required_columns = ['user_id', 'feature_1', 'feature_2', 'target'] assert all(col in df.columns for col in required_columns), \ "Missing required columns" # Data types assert df['user_id'].dtype == 'int64', "user_id must be integer" assert df['feature_1'].dtype == 'float64', "feature_1 must be float" # Value ranges assert df['feature_1'].min() >= 0, "feature_1 should be non-negative" assert df['feature_1'].max() <= 1000, "feature_1 exceeds upper bound" # Missing values assert df['user_id'].isna().sum() == 0, "user_id has missing values" assert df['target'].isna().sum() == 0, "target has missing values" # Duplicates assert not df.duplicated(subset=['user_id']).any(), \ "Duplicate user_ids found" return True # Using Great Expectations for formal validation suite = ge.build_suite( validators=[ ge.validators.ColumnValuesUnique(column='user_id'), ge.validators.ColumnValuesInSet( column='country', value_set=['US', 'CA', 'MX', 'UK'] ), ge.validators.ColumnValuesBetween( column='age', min_value=0, max_value=150 ) ] ) results = suite.validate(dataframe=df) assert results.success, "Data validation failed"

Model Testing

Example

import pytest from sklearn.metrics import f1_score, roc_auc_score, mean_squared_error import numpy as np class TestModelPerformance: """Test model meets minimum performance standards""" @pytest.fixture def model(self): return load_production_model() @pytest.fixture def test_data(self): X_test, y_test = load_test_data() return X_test, y_test def test_minimum_f1_score(self, model, test_data): """Ensure F1 score above minimum threshold""" X_test, y_test = test_data y_pred = model.predict(X_test) f1 = f1_score(y_test, y_pred) minimum_f1 = 0.75 assert f1 >= minimum_f1, \ f"F1 score {f1:.3f} below threshold {minimum_f1}" def test_auc_roc_performance(self, model, test_data): """Check AUC-ROC metric""" X_test, y_test = test_data y_proba = model.predict_proba(X_test)[:, 1] auc = roc_auc_score(y_test, y_proba) minimum_auc = 0.85 assert auc >= minimum_auc, \ f"AUC {auc:.3f} below threshold {minimum_auc}" def test_prediction_latency(self, model, test_data): """Ensure predictions fast enough for production""" import time X_test, _ = test_data start = time.time() predictions = model.predict(X_test) total_time = time.time() - start avg_latency_ms = (total_time / len(X_test)) * 1000 max_latency_ms = 100 # 100ms per prediction assert avg_latency_ms < max_latency_ms, \ f"Latency {avg_latency_ms:.1f}ms exceeds {max_latency_ms}ms" def test_model_robustness_missing_values(self, model, test_data): """Test model handles missing values gracefully""" X_test, _ = test_data # Add missing values X_with_missing = X_test.copy() X_with_missing.iloc[0, 0] = np.nan X_with_missing.iloc[1, 1] = np.nan # Should not raise exception predictions = model.predict(X_with_missing) assert len(predictions) == len(X_test) def test_model_robustness_outliers(self, model, test_data): """Test model with extreme outlier values""" X_test, _ = test_data # Add outliers X_with_outliers = X_test.copy() X_with_outliers.iloc[0, 0] = 1e10 # extreme value X_with_outliers.iloc[1, 1] = -1e10 predictions = model.predict(X_with_outliers) # Should produce reasonable predictions assert not np.isnan(predictions).any(), "NaN predictions" assert not np.isinf(predictions).any(), "Inf predictions" # Run tests with pytest # pytest tests/test_model.py -v

Integration Tests

Example

def test_end_to_end_pipeline(): """Test complete pipeline from raw data to predictions""" # 1. Load raw data raw_data = pd.read_csv('tests/test_data/raw_sample.csv') print(f"Loaded {len(raw_data)} samples") # 2. Validate raw data validate_input_data(raw_data) print("✓ Data validation passed") # 3. Preprocess data processed_data = preprocess_data(raw_data) assert len(processed_data) == len(raw_data), "Data lost during preprocessing" print("✓ Preprocessing completed") # 4. Generate features features = engineer_features(processed_data) assert features.shape[0] == len(raw_data), "Feature count mismatch" print("✓ Feature engineering completed") # 5. Load production model model = load_production_model() print("✓ Model loaded") # 6. Make predictions predictions = model.predict(features) assert len(predictions) == len(raw_data), "Prediction count mismatch" print(f"✓ Generated {len(predictions)} predictions") # 7. Validate output assert not predictions.isna().any(), "NaN predictions" assert not predictions.isinf().any(), "Inf predictions" if hasattr(model, 'predict_proba'): proba = model.predict_proba(features) assert np.allclose(proba.sum(axis=1), 1.0), "Probabilities don't sum to 1" print("✓ End-to-end test passed") return True

GitHub Actions CI/CD Pipeline

Example

name: ML CI/CD Pipeline on: push: branches: [main, develop] paths: - 'src/**' - 'tests/**' - 'requirements.txt' pull_request: branches: [main] jobs: # Stage 1: Validation & Testing test: runs-on: ubuntu-latest strategy: matrix: python-version: ['3.8', '3.9', '3.10'] steps: - uses: actions/checkout@v3 - name: Set up Python ${{ matrix.python-version }} uses: actions/setup-python@v4 with: python-version: ${{ matrix.python-version }} - name: Cache dependencies uses: actions/cache@v3 with: path: ~/.cache/pip key: ${{ runner.os }}-pip-${{ hashFiles('requirements.txt') }} - name: Install dependencies run: | pip install -r requirements.txt pip install pytest pytest-cov great-expectations - name: Lint code run: | pip install flake8 flake8 src/ --count --select=E9,F63,F7,F82 --show-source --statistics - name: Data validation run: python tests/test_data_validation.py - name: Unit tests run: pytest tests/unit/ -v --cov=src --cov-report=xml - name: Model tests run: pytest tests/model/ -v - name: Integration tests run: pytest tests/integration/ -v - name: Upload coverage uses: codecov/codecov-action@v3 with: files: ./coverage.xml # Stage 2: Train & Validate Model train: needs: test runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 with: python-version: '3.9' - name: Install dependencies run: | pip install -r requirements.txt pip install mlflow - name: Train model run: python scripts/train.py env: MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }} - name: Validate metrics run: python scripts/validate_metrics.py - name: Upload metrics artifact uses: actions/upload-artifact@v3 with: name: model-metrics path: metrics.json # Stage 3: Deploy to Staging deploy-staging: needs: train runs-on: ubuntu-latest if: github.ref == 'refs/heads/develop' steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 with: python-version: '3.9' - name: Deploy to staging run: python scripts/deploy.py --env staging env: AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }} AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }} - name: Run smoke tests on staging run: pytest tests/smoke/ -v env: MODEL_ENDPOINT: ${{ secrets.STAGING_ENDPOINT }} # Stage 4: Deploy to Production deploy-production: needs: train runs-on: ubuntu-latest if: github.ref == 'refs/heads/main' steps: - uses: actions/checkout@v3 - uses: actions/setup-python@v4 with: python-version: '3.9' - name: Deploy to production (canary 5%) run: python scripts/deploy.py --env production --canary 0.05 env: AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }} AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }} - name: Monitor for 1 hour run: python scripts/monitor.py --duration 3600 --alert-threshold 0.05 - name: Gradual rollout to 100% run: python scripts/deploy.py --env production --canary 1.0

Model Registry with MLflow

Example

import mlflow from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import f1_score, roc_auc_score def train_and_register_model(X_train, y_train, X_test, y_test): """Train model and register in MLflow""" mlflow.set_experiment("customer-churn-prediction") with mlflow.start_run(): # Log hyperparameters params = { "n_estimators": 100, "max_depth": 10, "min_samples_split": 5, "random_state": 42 } mlflow.log_params(params) # Train model model = RandomForestClassifier(**params) model.fit(X_train, y_train) # Evaluate y_pred = model.predict(X_test) y_proba = model.predict_proba(X_test)[:, 1] f1 = f1_score(y_test, y_pred) auc = roc_auc_score(y_test, y_proba) # Log metrics mlflow.log_metric("f1_score", f1) mlflow.log_metric("roc_auc", auc) # Log model mlflow.sklearn.log_model(model, "model") # Log artifacts import json metrics = {"f1": f1, "auc": auc} with open("metrics.json", "w") as f: json.dump(metrics, f) mlflow.log_artifact("metrics.json") run_id = mlflow.active_run().info.run_id # Register model if performance good if f1 >= 0.85 and auc >= 0.90: mlflow.register_model( f"runs:/{run_id}/model", "ChurnPredictionModel" ) print("✓ Model registered") return model, run_id # Transition model to production def promote_to_production(model_name, run_id): """Move model through stages: Staging → Production""" client = mlflow.tracking.MlflowClient() model_uri = f"runs:/{run_id}/model" # Register model version version = mlflow.register_model(model_uri, model_name) print(f"Registered model version {version.version}") # Move to staging client.transition_model_version_stage( name=model_name, version=version.version, stage="Staging" ) print("→ Transitioned to Staging") # After validation, move to production client.transition_model_version_stage( name=model_name, version=version.version, stage="Production" ) print("→ Transitioned to Production")

Deployment Strategies

Shadow Deployment

Run new model alongside current, log differences for analysis:

python example

class ShadowDeployment:
    def __init__(self, current_model, new_model):
        self.current_model = current_model
        self.new_model = new_model
        self.logger = setup_logger()
    
    def predict(self, X):
        """Use current model, shadow new model"""
        # Production: use current model
        current_pred = self.current_model.predict(X)
        
        # Experiment: run new model
        new_pred = self.new_model.predict(X)
        
        # Log comparison
        diff = np.abs(current_pred - new_pred).mean()
        self.logger.info(f"Shadow prediction difference: {diff:.4f}")
        
        return current_pred

Canary Release

Gradually increase traffic to new model:

python example

def canary_deployment(user_id, X):
    """Route percentage of traffic to new model"""
    
    traffic_percent = get_current_traffic_percent()  # 5%, 10%, 50%
    
    # Consistent user routing (hash-based)
    hash_value = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16)
    
    if (hash_value % 100) < traffic_percent:
        return new_model.predict(X), 'new'
    else:
        return current_model.predict(X), 'current'

Monitoring & Alerting

Example

import logging from datetime import datetime, timedelta class MLOpsMonitor: def __init__(self, alert_email): self.alert_email = alert_email self.logger = logging.getLogger(__name__) def monitor_model_performance(self, y_true, y_pred): """Alert if performance drops""" from sklearn.metrics import f1_score current_f1 = f1_score(y_true, y_pred) baseline_f1 = 0.85 if current_f1 < baseline_f1 * 0.95: # >5% drop self.send_alert( f"Performance degradation: F1 dropped to {current_f1:.3f}", "critical" ) def monitor_data_drift(self, X_new): """Alert if input distribution changed""" from scipy.stats import ks_2samp X_baseline = load_baseline_data() for col in X_new.columns: stat, p_value = ks_2samp(X_baseline[col], X_new[col]) if p_value < 0.05: self.send_alert( f"Data drift detected in {col}", "warning" ) def send_alert(self, message, severity): """Send alert email""" import smtplib subject = f"[{severity.upper()}] ML Model Alert" body = f"{datetime.now()}: {message}" # Send email logic here self.logger.log(severity, f"Alert: {message}")

Summary

CI/CD for ML requires:

Data validation
Model testing
Automated pipelines
Version control (code + models)
Model registry
Deployment strategies
Production monitoring
Alert triggers & retraining

Exam Focus

Revise definitions, diagrams, examples, and short-answer points for CI/CD for Machine Learning.

Interview Use

Prepare one clear explanation, one practical example, and one common mistake for this Machine Learning topic.

Search Terms

machine-learning, machine learning, machine, learning, mlops, for, ci/cd for machine learning

Continue learning this concept

MlopsDocker for Machine LearningContainerizing ML models and workflows using Docker for reproducibility, scalability, and deployment.MlopsFastAPI for Machine Learning APIsBuilding high-performance ML APIs with FastAPI, async support, and automatic documentation.MlopsFlask API for Machine LearningBuilding ML model serving APIs with Flask, request handling, and deployment.MlopsMLOps: IntroductionIntroduction to MLOps: productionizing ML systems, monitoring, deployment, and lifecycle management.MlopsModel Deployment StrategiesDifferent approaches to deploying ML models: batch, real-time, shadow, canary, blue-green, and A/B testing.MlopsML Model Monitoring & ObservabilityMonitoring ML models in production: performance metrics, data drift detection, alerting, and logging.

ML Notes

CI/CD for Machine Learning

Implementing continuous integration and continuous deployment pipelines for ML models including testing, validation, and automated deployment.

Why CI/CD for ML Matters

Traditional software CI/CD isn't enough for ML:

Code may not change but model performance can degrade (data drift)
Need to version data, features, models, and code
Testing must include data validation and model performance
Deployment requires A/B testing and gradual rollout

Example

Traditional Software: ML System: Code → Build → Test → Deploy Data + Code → Pipeline → Train → Test → Deploy ↑ Monitor, Detect Drift, Retrain ↓

ML Pipeline Architecture

Version Control Best Practices

What to Version Control

Example

Git Workflow for ML

python example

# 1. Create feature branch
git checkout -b feature/new-model-type

# 2. Experiment locally
# ... modify code, train models ...

# 3. Commit with clear messages
git commit -m "Implement gradient boosting model with feature engineering"

# 4. Push to trigger CI tests
git push origin feature/new-model-type

# 5. Create pull request
# - Runs automated tests
# - Checks performance vs baseline
# - Code review
# - Merge to main

# 6. CD automatically deploys after merge

Automated Testing for ML

Data Validation Tests

Example

Model Testing

Example

Integration Tests

Example

GitHub Actions CI/CD Pipeline

Example

Model Registry with MLflow

Example

Deployment Strategies

Shadow Deployment

Run new model alongside current, log differences for analysis:

python example

class ShadowDeployment:
    def __init__(self, current_model, new_model):
        self.current_model = current_model
        self.new_model = new_model
        self.logger = setup_logger()
    
    def predict(self, X):
        """Use current model, shadow new model"""
        # Production: use current model
        current_pred = self.current_model.predict(X)
        
        # Experiment: run new model
        new_pred = self.new_model.predict(X)
        
        # Log comparison
        diff = np.abs(current_pred - new_pred).mean()
        self.logger.info(f"Shadow prediction difference: {diff:.4f}")
        
        return current_pred

Canary Release

Gradually increase traffic to new model:

python example

def canary_deployment(user_id, X):
    """Route percentage of traffic to new model"""
    
    traffic_percent = get_current_traffic_percent()  # 5%, 10%, 50%
    
    # Consistent user routing (hash-based)
    hash_value = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16)
    
    if (hash_value % 100) < traffic_percent:
        return new_model.predict(X), 'new'
    else:
        return current_model.predict(X), 'current'

Monitoring & Alerting

Example

Summary

CI/CD for ML requires:

Data validation
Model testing
Automated pipelines
Version control (code + models)
Model registry
Deployment strategies
Production monitoring
Alert triggers & retraining

Exam Focus

Revise definitions, diagrams, examples, and short-answer points for CI/CD for Machine Learning.

Interview Use

Prepare one clear explanation, one practical example, and one common mistake for this Machine Learning topic.

Search Terms

machine-learning, machine learning, machine, learning, mlops, for, ci/cd for machine learning