Chapter 3: The ML Lifecycle

Introduction

The ML lifecycle encompasses all stages from problem definition to model retirement. Understanding this lifecycle is fundamental to implementing effective MLOps practices.

The Complete ML Lifecycle

Stage 1: Problem Definition

Before any ML work begins, clearly define the problem.

Key Questions to Answer

Question	Example Answer
What business problem are we solving?	Reduce customer churn by 15%
What type of ML problem is this?	Binary classification
What is the success metric?	AUC-ROC > 0.85
What are the latency requirements?	< 100ms for predictions
What data is available?	2 years of customer data

Problem Definition Document

# ml_problem_definition.yaml
project_name: Customer Churn Prediction
version: 1.0

business_context:
  objective: Predict customers likely to churn within 30 days
  impact: $2M annual revenue at risk
  stakeholders:
    - Customer Success Team
    - Marketing Team
    - Product Team

ml_formulation:
  task_type: binary_classification
  target_variable: churned_within_30_days
  positive_class_definition: "Customer cancelled subscription"
  
success_criteria:
  primary_metric: auc_roc
  threshold: 0.85
  secondary_metrics:
    - precision_at_10_percent_recall: 0.70
    - f1_score: 0.75

constraints:
  latency_p99_ms: 100
  model_size_mb: 500
  interpretability: high  # Need to explain predictions
  
timeline:
  mvp: 2024-03-01
  production: 2024-04-01

Stage 2: Data Collection

Gathering and understanding the data available for the ML project.

Data Inventory

# data_inventory.py
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

@dataclass
class DataSource:
    name: str
    location: str
    format: str
    owner: str
    update_frequency: str
    size_gb: float
    date_range_start: datetime
    date_range_end: datetime
    pii_columns: List[str]
    quality_score: float  # 0-1

# Define available data sources
data_sources = [
    DataSource(
        name="customer_transactions",
        location="s3://data-lake/transactions/",
        format="parquet",
        owner="data-engineering",
        update_frequency="daily",
        size_gb=150.0,
        date_range_start=datetime(2022, 1, 1),
        date_range_end=datetime(2024, 1, 1),
        pii_columns=["email", "phone"],
        quality_score=0.95
    ),
    DataSource(
        name="customer_support_tickets",
        location="s3://data-lake/support/",
        format="json",
        owner="customer-success",
        update_frequency="hourly",
        size_gb=25.0,
        date_range_start=datetime(2022, 6, 1),
        date_range_end=datetime(2024, 1, 1),
        pii_columns=["customer_email"],
        quality_score=0.88
    ),
]

Data Collection Pipeline

# data_collection_pipeline.py
from datetime import datetime, timedelta
import pandas as pd
from prefect import flow, task

@task
def extract_transactions(start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """Extract transaction data from data warehouse."""
    query = f"""
    SELECT 
        customer_id,
        transaction_date,
        amount,
        product_category
    FROM transactions
    WHERE transaction_date BETWEEN '{start_date}' AND '{end_date}'
    """
    return pd.read_sql(query, connection)

@task
def extract_support_tickets(start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """Extract support ticket data."""
    query = f"""
    SELECT 
        customer_id,
        ticket_date,
        category,
        resolution_time_hours,
        satisfaction_score
    FROM support_tickets
    WHERE ticket_date BETWEEN '{start_date}' AND '{end_date}'
    """
    return pd.read_sql(query, connection)

@task
def merge_data(transactions: pd.DataFrame, tickets: pd.DataFrame) -> pd.DataFrame:
    """Merge all data sources."""
    return transactions.merge(
        tickets, 
        on='customer_id', 
        how='left'
    )

@flow(name="data-collection")
def collect_training_data(lookback_days: int = 365):
    end_date = datetime.now()
    start_date = end_date - timedelta(days=lookback_days)
    
    transactions = extract_transactions(start_date, end_date)
    tickets = extract_support_tickets(start_date, end_date)
    
    merged = merge_data(transactions, tickets)
    
    return merged

Stage 3: Data Preparation

Cleaning and preparing data for feature engineering and training.

Data Quality Checks

# data_quality.py
import great_expectations as gx

def validate_training_data(df):
    """Validate training data quality."""
    context = gx.get_context()
    
    # Create expectations
    expectations = [
        # No null values in critical columns
        {"expectation_type": "expect_column_values_to_not_be_null",
         "kwargs": {"column": "customer_id"}},
        
        # Values in expected range
        {"expectation_type": "expect_column_values_to_be_between",
         "kwargs": {"column": "age", "min_value": 18, "max_value": 120}},
        
        # Unique identifiers
        {"expectation_type": "expect_column_values_to_be_unique",
         "kwargs": {"column": "customer_id"}},
        
        # No future dates
        {"expectation_type": "expect_column_values_to_be_between",
         "kwargs": {"column": "signup_date", 
                    "max_value": datetime.now().isoformat()}},
        
        # Column exists
        {"expectation_type": "expect_column_to_exist",
         "kwargs": {"column": "churned"}},
    ]
    
    # Run validation
    results = context.run_validation_operator(
        "action_list_operator",
        assets_to_validate=[df],
        run_id=f"training_data_{datetime.now().isoformat()}"
    )
    
    return results.success

Data Cleaning Pipeline

# data_cleaning.py
import pandas as pd
import numpy as np

def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """Clean and prepare raw data."""
    
    # Make a copy
    df = df.copy()
    
    # Handle missing values
    df['age'] = df['age'].fillna(df['age'].median())
    df['income'] = df['income'].fillna(df.groupby('segment')['income'].transform('median'))
    
    # Remove duplicates
    df = df.drop_duplicates(subset=['customer_id'], keep='last')
    
    # Handle outliers (clip to percentiles)
    for col in ['transaction_amount', 'login_frequency']:
        lower = df[col].quantile(0.01)
        upper = df[col].quantile(0.99)
        df[col] = df[col].clip(lower, upper)
    
    # Standardize text columns
    df['segment'] = df['segment'].str.lower().str.strip()
    
    # Convert data types
    df['signup_date'] = pd.to_datetime(df['signup_date'])
    df['churned'] = df['churned'].astype(int)
    
    return df

Stage 4: Feature Engineering

Creating features that help the model learn patterns.

Feature Engineering Pipeline

# feature_engineering.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, LabelEncoder
from datetime import datetime

class FeatureEngineer:
    def __init__(self):
        self.scaler = StandardScaler()
        self.encoders = {}
        
    def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Create all features for the model."""
        features = pd.DataFrame()
        features['customer_id'] = df['customer_id']
        
        # Temporal features
        features['account_age_days'] = (datetime.now() - df['signup_date']).dt.days
        features['days_since_last_login'] = (datetime.now() - df['last_login']).dt.days
        features['days_since_last_purchase'] = (datetime.now() - df['last_purchase']).dt.days
        
        # Aggregation features
        features['total_purchases'] = df['purchase_count']
        features['total_spend'] = df['total_spend']
        features['avg_order_value'] = df['total_spend'] / df['purchase_count'].clip(1)
        
        # Engagement features
        features['login_frequency_30d'] = df['logins_last_30d']
        features['support_tickets_90d'] = df['support_tickets_90d']
        features['nps_score'] = df['nps_score']
        
        # Ratio features
        features['spend_to_income_ratio'] = df['total_spend'] / df['income'].clip(1)
        features['engagement_trend'] = (
            df['logins_last_30d'] / df['logins_prev_30d'].clip(1)
        )
        
        # Categorical encoding
        for col in ['segment', 'plan_type', 'channel']:
            if col not in self.encoders:
                self.encoders[col] = LabelEncoder()
                features[f'{col}_encoded'] = self.encoders[col].fit_transform(df[col])
            else:
                features[f'{col}_encoded'] = self.encoders[col].transform(df[col])
        
        return features
    
    def scale_features(self, features: pd.DataFrame, fit: bool = False) -> pd.DataFrame:
        """Scale numerical features."""
        numerical_cols = features.select_dtypes(include=[np.number]).columns
        numerical_cols = [c for c in numerical_cols if c != 'customer_id']
        
        if fit:
            features[numerical_cols] = self.scaler.fit_transform(features[numerical_cols])
        else:
            features[numerical_cols] = self.scaler.transform(features[numerical_cols])
        
        return features

Stage 5: Model Training

Training and tuning models with proper experiment tracking.

Training Pipeline with MLflow

# training_pipeline.py
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import cross_val_score, GridSearchCV
from sklearn.metrics import roc_auc_score, precision_recall_curve
import pandas as pd
import numpy as np

class ModelTrainer:
    def __init__(self, experiment_name: str):
        mlflow.set_experiment(experiment_name)
        self.best_model = None
        self.best_score = 0
        
    def train_and_evaluate(
        self, 
        X_train: pd.DataFrame, 
        y_train: pd.Series,
        X_val: pd.DataFrame,
        y_val: pd.Series
    ):
        """Train multiple models and select the best one."""
        
        models = {
            'random_forest': {
                'model': RandomForestClassifier(random_state=42),
                'params': {
                    'n_estimators': [100, 200],
                    'max_depth': [10, 20, None],
                    'min_samples_split': [2, 5]
                }
            },
            'gradient_boosting': {
                'model': GradientBoostingClassifier(random_state=42),
                'params': {
                    'n_estimators': [100, 200],
                    'max_depth': [3, 5, 7],
                    'learning_rate': [0.01, 0.1]
                }
            }
        }
        
        for model_name, config in models.items():
            with mlflow.start_run(run_name=model_name):
                # Log parameters
                mlflow.log_param("model_type", model_name)
                
                # Grid search for best hyperparameters
                grid_search = GridSearchCV(
                    config['model'],
                    config['params'],
                    cv=5,
                    scoring='roc_auc',
                    n_jobs=-1
                )
                grid_search.fit(X_train, y_train)
                
                # Get best model
                best_estimator = grid_search.best_estimator_
                
                # Log best parameters
                for param, value in grid_search.best_params_.items():
                    mlflow.log_param(param, value)
                
                # Evaluate on validation set
                val_predictions = best_estimator.predict_proba(X_val)[:, 1]
                val_auc = roc_auc_score(y_val, val_predictions)
                
                # Log metrics
                mlflow.log_metric("val_auc_roc", val_auc)
                mlflow.log_metric("cv_score", grid_search.best_score_)
                
                # Log model
                mlflow.sklearn.log_model(best_estimator, "model")
                
                # Track best model
                if val_auc > self.best_score:
                    self.best_score = val_auc
                    self.best_model = best_estimator
                    
                print(f"{model_name}: AUC-ROC = {val_auc:.4f}")
        
        return self.best_model

Stage 6: Model Evaluation

Comprehensive evaluation before deployment.

Evaluation Framework

# model_evaluation.py
import matplotlib.pyplot as plt
from sklearn.metrics import (
    roc_auc_score, precision_recall_curve, 
    confusion_matrix, classification_report
)
import numpy as np

class ModelEvaluator:
    def __init__(self, model, X_test, y_test):
        self.model = model
        self.X_test = X_test
        self.y_test = y_test
        self.predictions = model.predict(X_test)
        self.probabilities = model.predict_proba(X_test)[:, 1]
        
    def generate_report(self) -> dict:
        """Generate comprehensive evaluation report."""
        
        report = {
            'auc_roc': roc_auc_score(self.y_test, self.probabilities),
            'classification_report': classification_report(
                self.y_test, 
                self.predictions, 
                output_dict=True
            ),
            'confusion_matrix': confusion_matrix(
                self.y_test, 
                self.predictions
            ).tolist()
        }
        
        # Calculate precision at different recall thresholds
        precision, recall, thresholds = precision_recall_curve(
            self.y_test, 
            self.probabilities
        )
        
        report['precision_at_recall_80'] = precision[
            np.argmin(np.abs(recall - 0.8))
        ]
        
        return report
    
    def check_deployment_criteria(self, criteria: dict) -> bool:
        """Check if model meets deployment criteria."""
        report = self.generate_report()
        
        checks = []
        for metric, threshold in criteria.items():
            if metric == 'auc_roc':
                passed = report['auc_roc'] >= threshold
            elif metric == 'precision':
                passed = report['classification_report']['1']['precision'] >= threshold
            elif metric == 'recall':
                passed = report['classification_report']['1']['recall'] >= threshold
            else:
                passed = True
                
            checks.append(passed)
            print(f"{metric}: {'PASS' if passed else 'FAIL'}")
        
        return all(checks)

Stage 7: Model Deployment

Deploying the model to production.

Deployment Configuration

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: churn-model
  labels:
    app: churn-prediction
spec:
  replicas: 3
  selector:
    matchLabels:
      app: churn-prediction
  template:
    metadata:
      labels:
        app: churn-prediction
    spec:
      containers:
      - name: model-server
        image: mlplatform/churn-model:v2.1.0
        ports:
        - containerPort: 8080
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        env:
        - name: MODEL_NAME
          value: "churn-model"
        - name: MODEL_VERSION
          value: "2.1.0"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5

Stage 8: Monitoring

Continuous monitoring of model performance.

Monitoring Metrics

# monitoring.py
from prometheus_client import Counter, Histogram, Gauge
from datetime import datetime
import numpy as np

# Define metrics
prediction_counter = Counter(
    'model_predictions_total',
    'Total predictions made',
    ['model_version', 'prediction_class']
)

prediction_latency = Histogram(
    'model_prediction_latency_seconds',
    'Prediction latency',
    buckets=[0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5]
)

data_drift_gauge = Gauge(
    'model_data_drift_score',
    'Data drift score by feature',
    ['feature_name']
)

model_accuracy_gauge = Gauge(
    'model_accuracy_rolling',
    'Rolling accuracy over last 24 hours'
)

class ModelMonitor:
    def __init__(self, model_version: str):
        self.model_version = model_version
        self.predictions_log = []
        
    def log_prediction(self, input_data: dict, prediction: int, latency: float):
        """Log a prediction for monitoring."""
        prediction_counter.labels(
            model_version=self.model_version,
            prediction_class=str(prediction)
        ).inc()
        
        prediction_latency.observe(latency)
        
        self.predictions_log.append({
            'timestamp': datetime.now(),
            'input': input_data,
            'prediction': prediction,
            'latency': latency
        })
    
    def check_drift(self, current_data: np.ndarray, reference_data: np.ndarray):
        """Check for data drift using statistical tests."""
        from scipy import stats
        
        drift_scores = {}
        for i, feature in enumerate(feature_names):
            # Kolmogorov-Smirnov test
            statistic, p_value = stats.ks_2samp(
                reference_data[:, i],
                current_data[:, i]
            )
            drift_scores[feature] = statistic
            data_drift_gauge.labels(feature_name=feature).set(statistic)
        
        return drift_scores

Stage 9: Retraining

Triggering model retraining when needed.

Retraining Triggers

# retraining_triggers.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from datetime import datetime, timedelta

class TriggerType(Enum):
    SCHEDULED = "scheduled"
    PERFORMANCE = "performance"
    DRIFT = "drift"
    DATA_VOLUME = "data_volume"
    MANUAL = "manual"

@dataclass
class RetrainingTrigger:
    trigger_type: TriggerType
    threshold: Optional[float] = None
    schedule: Optional[str] = None
    
def check_retraining_needed(
    current_auc: float,
    baseline_auc: float,
    drift_score: float,
    last_training_date: datetime,
    new_data_count: int
) -> tuple[bool, TriggerType]:
    """Check if retraining is needed."""
    
    triggers = [
        # Performance degradation
        RetrainingTrigger(
            trigger_type=TriggerType.PERFORMANCE,
            threshold=0.05  # 5% AUC drop
        ),
        # Data drift
        RetrainingTrigger(
            trigger_type=TriggerType.DRIFT,
            threshold=0.1  # Drift score > 0.1
        ),
        # Scheduled retraining
        RetrainingTrigger(
            trigger_type=TriggerType.SCHEDULED,
            schedule="30 days"
        ),
        # New data volume
        RetrainingTrigger(
            trigger_type=TriggerType.DATA_VOLUME,
            threshold=100000  # 100k new records
        ),
    ]
    
    # Check performance
    if baseline_auc - current_auc > 0.05:
        return True, TriggerType.PERFORMANCE
    
    # Check drift
    if drift_score > 0.1:
        return True, TriggerType.DRIFT
    
    # Check schedule
    if datetime.now() - last_training_date > timedelta(days=30):
        return True, TriggerType.SCHEDULED
    
    # Check data volume
    if new_data_count > 100000:
        return True, TriggerType.DATA_VOLUME
    
    return False, None

Key Takeaways

Summary

1. Problem definition is crucial - unclear goals lead to failed projects
2. Data quality determines model quality - garbage in, garbage out
3. Feature engineering often matters more than model selection
4. Evaluation must be comprehensive and match business metrics
5. Monitoring is not optional - models degrade without it
6. Retraining keeps models relevant as data evolves

Further Reading

• Google's Rules of Machine Learning
• Uber's Michelangelo Platform
• Netflix's ML Infrastructure

---

Next: Module 1 Quiz