Production MLOps: From Jupyter Notebook to Scalable Model Serving

2025-04-05

Building an end-to-end MLOps pipeline with Kubeflow, MLflow, Seldon Core, and automated drift detection for production ML systems

The gap between training a model in a Jupyter notebook and serving it reliably in production is enormous. Most ML models never make it to production — not because they don't perform well, but because organizations lack the infrastructure to deploy, monitor, and maintain them at scale.

This guide covers building a production MLOps pipeline that serves 15+ models with sub-hour deployment times and automated drift detection.

Production MLOps Pipeline Architecture

The MLOps Maturity Gap

Here's what most ML teams deal with:

Level 0: Data scientists train models in notebooks, hand off pickled files to engineers
Level 1: Basic CI/CD for model code, manual deployment
Level 2: Automated training pipelines, model registry, basic monitoring
Level 3: Full automation — continuous training, serving, monitoring, and retraining

Most organizations are stuck between Level 0 and Level 1. Our pipeline operates at Level 3.

Pipeline Architecture

Stage 1: Data Versioning with DVC

Every ML experiment must be reproducible. DVC (Data Version Control) tracks datasets alongside code:

# dvc.yaml - Pipeline definition
stages:
  prepare:
    cmd: python src/prepare.py
    deps:
      - src/prepare.py
      - data/raw/
    params:
      - prepare.split_ratio
      - prepare.seed
    outs:
      - data/processed/train.parquet
      - data/processed/test.parquet

  train:
    cmd: python src/train.py
    deps:
      - src/train.py
      - data/processed/train.parquet
    params:
      - train.learning_rate
      - train.epochs
      - train.batch_size
    outs:
      - models/model.onnx
    metrics:
      - metrics/train_metrics.json:
          cache: false

  evaluate:
    cmd: python src/evaluate.py
    deps:
      - src/evaluate.py
      - models/model.onnx
      - data/processed/test.parquet
    metrics:
      - metrics/eval_metrics.json:
          cache: false
    plots:
      - metrics/confusion_matrix.csv

Stage 2: Training Orchestration with Kubeflow

Kubeflow Pipelines provides scalable, reproducible training on Kubernetes:

# pipelines/training_pipeline.py
from kfp import dsl, compiler
from kfp.dsl import Input, Output, Dataset, Model, Metrics

@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas", "scikit-learn", "mlflow"]
)
def train_model(
    train_data: Input[Dataset],
    hyperparams: dict,
    model_artifact: Output[Model],
    metrics: Output[Metrics],
):
    import pandas as pd
    from sklearn.ensemble import GradientBoostingClassifier
    from sklearn.metrics import accuracy_score, f1_score
    import mlflow
    import joblib

    # Load data
    df = pd.read_parquet(train_data.path)
    X = df.drop('target', axis=1)
    y = df['target']

    # Train with hyperparameters
    model = GradientBoostingClassifier(
        learning_rate=hyperparams['learning_rate'],
        n_estimators=hyperparams['n_estimators'],
        max_depth=hyperparams['max_depth'],
    )
    model.fit(X, y)

    # Log metrics
    predictions = model.predict(X)
    accuracy = accuracy_score(y, predictions)
    f1 = f1_score(y, predictions, average='weighted')

    metrics.log_metric('accuracy', accuracy)
    metrics.log_metric('f1_score', f1)

    # Save model
    joblib.dump(model, model_artifact.path)

    # Log to MLflow
    with mlflow.start_run():
        mlflow.log_params(hyperparams)
        mlflow.log_metrics({'accuracy': accuracy, 'f1': f1})
        mlflow.sklearn.log_model(model, "model")


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["evidently", "pandas"]
)
def validate_model(
    model: Input[Model],
    test_data: Input[Dataset],
    baseline_data: Input[Dataset],
    validation_result: Output[Metrics],
) -> bool:
    """Validate model against quality gates before promotion."""
    import pandas as pd
    from evidently.test_suite import TestSuite
    from evidently.tests import (
        TestAccuracyScore,
        TestF1Score,
        TestColumnDrift,
    )

    current = pd.read_parquet(test_data.path)
    reference = pd.read_parquet(baseline_data.path)

    # Run data quality and model performance tests
    suite = TestSuite(tests=[
        TestAccuracyScore(gte=0.85),  # Must beat threshold
        TestF1Score(gte=0.80),
        TestColumnDrift(column_name='prediction'),
    ])

    suite.run(reference_data=reference, current_data=current)
    results = suite.as_dict()

    passed = all(test['status'] == 'SUCCESS' for test in results['tests'])
    validation_result.log_metric('validation_passed', int(passed))

    return passed


@dsl.pipeline(name="ml-training-pipeline")
def training_pipeline(
    dataset_path: str,
    hyperparams: dict = {
        'learning_rate': 0.1,
        'n_estimators': 200,
        'max_depth': 5,
    },
):
    # Step 1: Prepare data
    prepare_task = prepare_data(dataset_path=dataset_path)

    # Step 2: Train model
    train_task = train_model(
        train_data=prepare_task.outputs['train_data'],
        hyperparams=hyperparams,
    )

    # Step 3: Validate model
    validate_task = validate_model(
        model=train_task.outputs['model_artifact'],
        test_data=prepare_task.outputs['test_data'],
        baseline_data=prepare_task.outputs['baseline_data'],
    )

    # Step 4: Deploy only if validation passes
    with dsl.Condition(validate_task.output == True):
        deploy_model(
            model=train_task.outputs['model_artifact'],
            model_name="fraud-detector",
            namespace="ml-serving",
        )


# Compile and submit
compiler.Compiler().compile(training_pipeline, 'pipeline.yaml')

Stage 3: Model Serving with Seldon Core

Seldon Core provides production-grade model serving with A/B testing and canary deployments:

# seldon-deployment.yaml
apiVersion: machinelearning.seldon.io/v1
kind: SeldonDeployment
metadata:
  name: fraud-detector
  namespace: ml-serving
spec:
  predictors:
    # Canary: New model version (10% traffic)
    - name: canary
      traffic: 10
      componentSpecs:
        - spec:
            containers:
              - name: model
                image: ghcr.io/myorg/fraud-model:v2.1.0
                resources:
                  requests:
                    cpu: "2"
                    memory: "4Gi"
                  limits:
                    cpu: "4"
                    memory: "8Gi"
      graph:
        name: model
        type: MODEL
        implementation: SKLEARN_SERVER
        modelUri: s3://models/fraud-detector/v2.1.0
        parameters:
          - name: method
            type: STRING
            value: predict_proba

    # Default: Current stable version (90% traffic)
    - name: default
      traffic: 90
      componentSpecs:
        - spec:
            containers:
              - name: model
                image: ghcr.io/myorg/fraud-model:v2.0.0
                resources:
                  requests:
                    cpu: "2"
                    memory: "4Gi"
      graph:
        name: model
        type: MODEL
        implementation: SKLEARN_SERVER
        modelUri: s3://models/fraud-detector/v2.0.0

  # A/B test configuration
  annotations:
    seldon.io/ab-test: "true"
    seldon.io/canary-promote-threshold: "0.95" # Promote if 95%+ accuracy

Stage 4: Drift Detection and Auto-Retraining

Models degrade over time as data distributions shift. We detect this automatically:

# monitoring/drift_detector.py
from evidently.metrics import (
    DataDriftTable,
    DatasetDriftMetric,
    ColumnDriftMetric,
)
from evidently.report import Report
from prometheus_client import Gauge, start_http_server
import schedule
import time

# Prometheus metrics for Grafana dashboards
drift_score = Gauge(
    'model_data_drift_score',
    'Current data drift score',
    ['model_name', 'feature']
)
prediction_drift = Gauge(
    'model_prediction_drift',
    'Prediction distribution drift',
    ['model_name']
)

def check_drift(model_name: str, reference_data, current_data):
    """Run drift detection and expose metrics."""
    report = Report(metrics=[
        DatasetDriftMetric(),
        DataDriftTable(),
    ])

    report.run(
        reference_data=reference_data,
        current_data=current_data,
    )

    results = report.as_dict()

    # Update Prometheus metrics
    dataset_drift = results['metrics'][0]['result']['dataset_drift']
    drift_share = results['metrics'][0]['result']['drift_share']

    prediction_drift.labels(model_name=model_name).set(drift_share)

    # Per-feature drift scores
    for col_name, col_data in results['metrics'][1]['result']['drift_by_columns'].items():
        drift_score.labels(
            model_name=model_name,
            feature=col_name,
        ).set(col_data['drift_score'])

    # Trigger retraining if drift exceeds threshold
    if dataset_drift or drift_share > 0.3:
        trigger_retraining(model_name)

    return dataset_drift


def trigger_retraining(model_name: str):
    """Trigger Kubeflow pipeline for model retraining."""
    import kfp
    client = kfp.Client(host='http://kubeflow-pipelines:8888')
    client.create_run_from_pipeline_package(
        'pipeline.yaml',
        arguments={'dataset_path': f's3://data/{model_name}/latest/'},
        run_name=f'auto-retrain-{model_name}-{int(time.time())}',
    )


# Run drift check every 5 minutes
schedule.every(5).minutes.do(
    check_drift,
    model_name='fraud-detector',
    reference_data=load_reference_data(),
    current_data=load_current_data(),
)

GitOps for ML Models

Every model deployment is tracked in Git:

# argocd/ml-models/fraud-detector/kustomization.yaml
apiVersion: kustomize.config.k8s.io/v1beta1
kind: Kustomization
resources:
  - seldon-deployment.yaml
  - hpa.yaml
  - monitoring.yaml

images:
  - name: ghcr.io/myorg/fraud-model
    newTag: v2.1.0 # Updated by CI on model promotion

patches:
  - target:
      kind: SeldonDeployment
      name: fraud-detector
    patch: |
      - op: replace
        path: /spec/predictors/0/traffic
        value: 10  # Canary percentage, gradually increased

Results

Metric	Before MLOps	After MLOps
Model deployment time	3-5 days	< 1 hour
Models in production	3	15+
Drift detection time	Never (manual)	< 5 minutes
Experiment reproducibility	0%	100%
A/B test capability	None	All models
Rollback time	Hours (manual)	< 2 minutes (GitOps)

Lessons Learned

Version everything: Data, code, models, and infrastructure. If you can't reproduce it, you can't debug it.
Separate model training from serving: Training is batch-oriented and GPU-heavy. Serving is latency-sensitive and CPU-heavy. Different scaling patterns require different infrastructure.
Drift detection is non-negotiable: A model that worked great at deployment can silently degrade. Statistical tests on incoming data distributions are your early warning system.
Canary deployments for models: Never switch 100% traffic to a new model version. Start at 5-10%, monitor key metrics, and gradually promote.
Treat ML infrastructure as a platform: Data scientists shouldn't need to know Kubernetes. Provide simple abstractions (submit training job, deploy model, check metrics) that hide infrastructure complexity.

The MLOps Maturity Roadmap

Start here and progress iteratively:

Foundation (Month 1-2): MLflow for experiment tracking, DVC for data versioning
Automation (Month 3-4): Kubeflow pipelines for training, model registry for promotion
Serving (Month 5-6): Seldon Core with canary deployments and A/B testing
Monitoring (Month 7+): Drift detection, auto-retraining, SLOs for model performance

The goal isn't to automate everything at once — it's to systematically remove the friction between "model works in notebook" and "model works in production."