ChatWhole Learn — Free Programming & Data Science Tutorials

ChatWhole Learn

MLOps in 2026: End-to-End ML Pipeline Architecture

Machine learning without MLOps is like software without DevOps — it works in the lab but fails in production. This guide covers the complete MLOps stack, from feature engineering through model deployment and monitoring.

The MLOps Stack

Feature Store Architecture

Core Concepts

python

from feast import FeatureStore, Entity, Feature, ValueType
from feast import FileSource, FeatureView
from datetime import timedelta

# Define entities
user_entity = Entity(
    name="user_id",
    value_type=ValueType.INT64,
    description="User identifier"
)

# Define feature views
user_features = FeatureView(
    name="user_features",
    entities=["user_id"],
    ttl=timedelta(days=1),
    features=[
        Feature(name="age", dtype=ValueType.INT32),
        Feature(name="total_purchases", dtype=ValueType.FLOAT),
        Feature(name="avg_order_value", dtype=ValueType.FLOAT),
        Feature(name="days_since_last_purchase", dtype=ValueType.INT32),
    ],
    source=FileSource(
        path="s3://features/user_features.parquet",
        event_timestamp_column="event_timestamp"
    )
)

# Store features
store = FeatureStore(repo_path=".")
store.apply([user_entity, user_features])

# Retrieve features for training
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_features:age",
        "user_features:total_purchases",
        "user_features:avg_order_value",
    ]
).to_df()

# Retrieve features for serving (real-time)
feature_vector = store.get_online_features(
    features=[
        "user_features:age",
        "user_features:total_purchases",
        "user_features:avg_order_value",
    ],
    entity_rows=[{"user_id": 123}]
).to_dict()

Feature Store Best Practices

python

class FeatureStoreManager:
    def __init__(self, store: FeatureStore):
        self.store = store
    
    async def validate_features(self, feature_view: str):
        """Validate feature freshness and quality"""
        stats = await self.store.get_feature_statistics(feature_view)
        
        checks = {
            'freshness': stats.max_age < timedelta(hours=1),
            'completeness': stats.null_rate < 0.05,
            'distribution': self._check_distribution(stats),
            'drift': self._check_drift(stats),
        }
        
        return FeatureValidationResult(
            feature_view=feature_view,
            passed=all(checks.values()),
            details=checks
        )
    
    def _check_distribution(self, stats):
        """Check if feature distribution is within expected bounds"""
        for feature in stats.features:
            if feature.mean < feature.expected_mean * 0.5:
                return False
            if feature.mean > feature.expected_mean * 2.0:
                return False
        return True

Model Registry

MLflow Model Registry

python

import mlflow
import mlflow.sklearn
from mlflow.tracking import MlflowClient

# Train model
model = train_model(X_train, y_train)

# Log model
with mlflow.start_run():
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", 10)
    
    # Log metrics
    mlflow.log_metric("accuracy", accuracy)
    mlflow.log_metric("f1_score", f1)
    
    # Log model
    mlflow.sklearn.log_model(
        model,
        artifact_path="model",
        registered_model_name="churn_predictor"
    )

# Promote model to production
client = MlflowClient()
client.transition_model_version_stage(
    name="churn_predictor",
    version=1,
    stage="Production"
)

# Load production model
model = mlflow.pyfunc.load_model(
    "models:/churn_predictor/Production"
)

Model Versioning Strategy

python

class ModelVersionManager:
    def __init__(self, registry):
        self.registry = registry
    
    async def promote_model(self, model_name: str, version: int):
        """Promote model through stages with validation"""
        
        # Stage 1: Validation
        model = self.registry.load(model_name, version)
        val_results = await self.validate_model(model)
        
        if not val_results.passed:
            raise ModelValidationError(val_results.errors)
        
        # Stage 2: Shadow deployment
        await self.shadow_deploy(model_name, version)
        
        # Stage 3: A/B test
        ab_results = await self.run_ab_test(model_name, version)
        
        if ab_results.improvement < 0.01:  # Less than 1% improvement
            raise ModelNotImprovedError()
        
        # Stage 4: Full promotion
        self.registry.promote(model_name, version, "Production")
        
        return PromotionResult(
            model=model_name,
            version=version,
            improvement=ab_results.improvement
        )

ML Pipeline Orchestration

Airflow DAG for ML Pipeline

python

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.sensors.external_task import ExternalTaskSensor
from datetime import datetime, timedelta

default_args = {
    'owner': 'ml-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['[email protected]'],
    'retries': 2,
    'retry_delay': timedelta(minutes=5),
}

with DAG(
    'ml_training_pipeline',
    default_args=default_args,
    description='End-to-end ML training pipeline',
    schedule_interval='@weekly',
    start_date=datetime(2026, 1, 1),
    catchup=False,
) as dag:
    
    # Task 1: Extract and validate data
    extract_data = PythonOperator(
        task_id='extract_data',
        python_callable=extract_training_data,
        params={'date': '{{ ds }}'}
    )
    
    # Task 2: Feature engineering
    engineer_features = PythonOperator(
        task_id='engineer_features',
        python_callable=compute_features,
    )
    
    # Task 3: Train model
    train_model = PythonOperator(
        task_id='train_model',
        python_callable=train_and_validate,
    )
    
    # Task 4: Evaluate model
    evaluate_model = PythonOperator(
        task_id='evaluate_model',
        python_callable=evaluate_and_compare,
    )
    
    # Task 5: Deploy model
    deploy_model = PythonOperator(
        task_id='deploy_model',
        python_callable=deploy_to_production,
    )
    
    # Task 6: Monitor model
    setup_monitoring = PythonOperator(
        task_id='setup_monitoring',
        python_callable=configure_monitoring,
    )
    
    # Define dependencies
    extract_data >> engineer_features >> train_model
    train_model >> evaluate_model >> deploy_model
    deploy_model >> setup_monitoring

Dagster Pipeline

python

from dagster import op, job, Out, In
import pandas as pd

@op(out=Out(pd.DataFrame))
def extract_data(context):
    """Extract training data from data warehouse"""
    query = """
    SELECT * FROM features
    WHERE event_date = DATE '{{ ds }}'
    """
    return pd.read_sql(query, connection)

@op(ins={"data": In(pd.DataFrame)}, out=Out(pd.DataFrame))
def validate_data(context, data):
    """Validate data quality"""
    assert data.isnull().sum().sum() / data.size < 0.05
    assert len(data) > 1000
    return data

@op(ins={"data": In(pd.DataFrame)}, out=Out(dict))
def train_model(context, data):
    """Train ML model"""
    from sklearn.ensemble import RandomForestClassifier
    
    X = data.drop('target', axis=1)
    y = data['target']
    
    model = RandomForestClassifier(n_estimators=100)
    model.fit(X, y)
    
    return {"model": model, "features": X.columns.tolist()}

@job
def ml_pipeline():
    data = extract_data()
    validated = validate_data(data)
    result = train_model(validated)

Model Monitoring

Data Drift Detection

python

from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import (
    DataDriftPreset,
    TargetDriftPreset,
    ClassificationPreset
)

class ModelMonitor:
    def __init__(self, reference_data, production_data):
        self.reference = reference_data
        self.production = production_data
    
    def check_drift(self) -> DriftReport:
        """Comprehensive drift detection"""
        
        # Data drift
        data_drift_report = Report(metrics=[
            DataDriftPreset(),
        ])
        data_drift_report.run(
            current_data=self.production,
            reference_data=self.reference
        )
        
        # Model performance drift
        performance_report = Report(metrics=[
            ClassificationPreset(),
        ])
        performance_report.run(
            current_data=self.production,
            reference_data=self.reference
        )
        
        return DriftReport(
            data_drift=data_drift_report,
            performance_drift=performance_report,
            has_drift=self._detect_drift(data_drift_report)
        )
    
    def _detect_drift(self, report) -> bool:
        """Detect if significant drift occurred"""
        result = report.as_dict()
        for metric in result['metrics']:
            if metric.get('result', {}).get('drift_detected'):
                return True
        return False

Real-Time Monitoring Dashboard

python

class MonitoringDashboard:
    def __init__(self, metrics_store):
        self.metrics_store = metrics_store
    
    async def get_dashboard_data(self, model_name: str, hours: int = 24):
        """Get monitoring data for dashboard"""
        
        metrics = await self.metrics_store.get_metrics(
            model_name=model_name,
            start_time=datetime.now() - timedelta(hours=hours)
        )
        
        return {
            'latency': {
                'p50': np.percentile(metrics['latencies'], 50),
                'p95': np.percentile(metrics['latencies'], 95),
                'p99': np.percentile(metrics['latencies'], 99),
            },
            'throughput': {
                'requests_per_minute': len(metrics['latencies']) / (hours * 60),
                'total_requests': len(metrics['latencies']),
            },
            'errors': {
                'error_rate': metrics['errors'] / len(metrics['latencies']),
                'timeout_rate': metrics['timeouts'] / len(metrics['latencies']),
            },
            'quality': {
                'prediction_distribution': self._compute_distribution(
                    metrics['predictions']
                ),
                'feature_drift_scores': metrics.get('drift_scores', {}),
            }
        }

CI/CD for ML

GitHub Actions Pipeline

# .github/workflows/ml-pipeline.yml
name: ML Pipeline

on:
  push:
    branches: [main]
  schedule:
    - cron: '0 0 * * 0'  # Weekly

jobs:
  train:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run tests
        run: pytest tests/
      
      - name: Train model
        run: python train.py
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_URI }}
      
      - name: Evaluate model
        run: python evaluate.py
      
      - name: Deploy if approved
        if: success()
        run: python deploy.py

Model Testing

python

import pytest
import numpy as np

class TestModel:
    def test_prediction_shape(self, model):
        """Model returns correct output shape"""
        X = np.random.randn(10, 5)
        predictions = model.predict(X)
        assert predictions.shape == (10,)
    
    def test_prediction_range(self, model):
        """Predictions are in valid range"""
        X = np.random.randn(100, 5)
        predictions = model.predict(X)
        assert predictions.min() >= 0
        assert predictions.max() <= 1
    
    def test_latency(self, model):
        """Model inference is fast enough"""
        X = np.random.randn(1, 5)
        start = time.time()
        for _ in range(100):
            model.predict(X)
        latency = (time.time() - start) / 100
        assert latency < 0.1  # Less than 100ms
    
    def test_robustness(self, model):
        """Model handles edge cases"""
        # Empty input
        with pytest.raises(ValueError):
            model.predict(np.array([]))
        
        # NaN input
        X_nan = np.full((1, 5), np.nan)
        predictions = model.predict(X_nan)
        assert not np.any(np.isnan(predictions))

Production Architecture

Kubernetes Deployment

# model-serving.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: model-server
spec:
  replicas: 3
  selector:
    matchLabels:
      app: model-server
  template:
    metadata:
      labels:
        app: model-server
    spec:
      containers:
      - name: model
        image: model-server:latest
        resources:
          requests:
            memory: "4Gi"
            cpu: "2"
            nvidia.com/gpu: "1"
          limits:
            memory: "8Gi"
            cpu: "4"
            nvidia.com/gpu: "1"
        ports:
        - containerPort: 8080
        env:
        - name: MODEL_PATH
          value: "s3://models/production/model.pkl"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: model-server-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: model-server
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

Cost Optimization

Model Optimization Techniques

python

class ModelOptimizer:
    def quantize(self, model, precision="int8"):
        """Reduce model size with quantization"""
        if precision == "int8":
            return torch.quantization.quantize_dynamic(
                model, {torch.nn.Linear}, dtype=torch.qint8
            )
        elif precision == "fp16":
            return model.half()
    
    def prune(self, model, amount=0.3):
        """Remove unnecessary weights"""
        import torch.nn.utils.prune as prune
        for module in model.modules():
            if isinstance(module, torch.nn.Linear):
                prune.l1_unstructured(module, name='weight', amount=amount)
        return model
    
    def distill(self, teacher_model, student_model, train_data):
        """Knowledge distillation"""
        # Train student to mimic teacher
        for batch in train_data:
            with torch.no_grad():
                teacher_logits = teacher_model(batch)
            student_logits = student_model(batch)
            loss = kl_divergence(student_logits, teacher_logits)
            loss.backward()

Conclusion

MLOps in 2026 requires a comprehensive approach covering the entire ML lifecycle:

Feature Store — Centralized feature management with validation
Model Registry — Version control and staged promotion
Pipeline Orchestration — Automated training and deployment
Monitoring — Real-time drift detection and alerting
CI/CD — Automated testing and deployment

The key is to start simple and add complexity as needed. Begin with MLflow for tracking, Airflow for orchestration, and Evidently for monitoring. Scale up as your ML operations mature.

MLOps in 2026: End-to-End ML Pipeline Architecture

MLOps in 2026: End-to-End ML Pipeline Architecture

The MLOps Stack

Feature Store Architecture

Core Concepts

Feature Store Best Practices

Model Registry

MLflow Model Registry

Model Versioning Strategy

ML Pipeline Orchestration

Airflow DAG for ML Pipeline

Dagster Pipeline

Model Monitoring

Data Drift Detection

Real-Time Monitoring Dashboard

CI/CD for ML

GitHub Actions Pipeline

Model Testing

Production Architecture

Kubernetes Deployment

Cost Optimization

Model Optimization Techniques

Conclusion

Explore More Articles