Kubeflow Pipelines ML Workflow Orchestration Practical Guide: From KFP v2 SDK to Production Deployment

• Introduction
• Kubeflow Pipelines Architecture
  • Core Component Structure
  • Architecture Changes in KFP v2
• KFP v2 SDK Basics
  • Writing a Component
  • Custom Container Components
  • Defining the Pipeline
• Advanced Pipeline Patterns
  • Parallel Execution and Conditional Branching
  • Recurring Runs and the Exit Handler
• Pipeline Caching and Artifact Management
  • Caching Strategy
  • Artifact Types and Management
• Comparing Workflow Orchestration Tools
  • Criteria for Choosing
• A Practical Multi-Step ML Pipeline Example
  • The Full Pipeline: From Data Preparation to Deployment
• Managing Kubernetes Resources
  • Pod Resources and Node Affinity
• Operational Considerations
  • Resource Cautions
  • Security Cautions
• Failure Cases and Recovery Procedures
  • Case 1: Pod OOMKilled
  • Case 2: Pipeline Version Conflict
  • Case 3: Metadata DB Connection Failure
• Production Checklist
  • Infrastructure Setup
  • Pipeline Development
  • Operations and Monitoring
  • Security
• References

Introduction

As an ML project grows to production scale, reliably managing the whole workflow - from data preprocessing through model training, evaluation, and deployment - becomes the central challenge. Experiments in a Jupyter Notebook are hard to reproduce, and running scripts by hand is error-prone. Solving that calls for an ML pipeline orchestration tool.

Kubeflow Pipelines (KFP) is an open-source project led by Google: a platform for defining and running ML workflows on top of Kubernetes. It runs each step as an independent container, guaranteeing reproducibility, and supports pipeline version management and experiment tracking. This article covers the KFP v2 SDK in detail, from its architecture through building a practical pipeline to production operating strategy.

Kubeflow Pipelines Architecture

Core Component Structure

Kubeflow Pipelines is made up of several microservices.

Architecture Changes in KFP v2

KFP v2 brought fundamental architectural changes over v1. It removed the dependency on Argo Workflows and introduced its own workflow engine.

# Key differences between KFP v2 and v1
"""
KFP v1:
- Execution based on Argo Workflows
- Uses kfp.dsl.ContainerOp
- Pipelines can be defined in YAML

KFP v2:
- Its own workflow engine (or Argo, optionally)
- Uses the kfp.dsl.component decorator
- Introduces IR (Intermediate Representation) YAML
- Native ML Metadata integration
- Type-safe component interfaces
"""

# v2 architecture layers
ARCHITECTURE_LAYERS = {
    "SDK Layer": "Define the pipeline in Python DSL (kfp.dsl)",
    "IR Layer": "Platform-independent intermediate representation (PipelineSpec YAML)",
    "Backend Layer": "Pipeline execution and management (API Server)",
    "Runtime Layer": "Container orchestration (K8s Pod)",
    "Metadata Layer": "Execution history and artifact tracking (MLMD)",
}

KFP v2 SDK Basics

Writing a Component

In KFP v2 a component is defined with the @component decorator. Each component runs in its own independent container.

from kfp import dsl
from kfp.dsl import Input, Output, Dataset, Model, Metrics

# Lightweight Python component (when dependencies are few)
@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.1.4", "scikit-learn==1.4.0"],
)
def preprocess_data(
    raw_data_path: str,
    test_size: float,
    train_dataset: Output[Dataset],
    test_dataset: Output[Dataset],
    metrics: Output[Metrics],
):
    """Data preprocessing component"""
    import pandas as pd
    from sklearn.model_selection import train_test_split

    df = pd.read_csv(raw_data_path)

    # Handle missing values
    df = df.dropna(subset=["target"])
    df = df.fillna(df.median(numeric_only=True))

    # Train/test split
    train_df, test_df = train_test_split(
        df, test_size=test_size, random_state=42, stratify=df["target"]
    )

    # Save as an artifact
    train_df.to_csv(train_dataset.path, index=False)
    test_df.to_csv(test_dataset.path, index=False)

    # Log the metrics
    metrics.log_metric("total_samples", len(df))
    metrics.log_metric("train_samples", len(train_df))
    metrics.log_metric("test_samples", len(test_df))
    metrics.log_metric("feature_count", len(df.columns) - 1)

Custom Container Components

When you need heavy dependencies, use a custom container image.

# Component based on a custom image
@dsl.component(
    base_image="gcr.io/my-project/ml-training:v2.1",
)
def train_model(
    train_dataset: Input[Dataset],
    model_type: str,
    hyperparameters: dict,
    trained_model: Output[Model],
    metrics: Output[Metrics],
):
    """Model training component"""
    import pandas as pd
    import joblib
    from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
    from sklearn.metrics import accuracy_score, f1_score

    train_df = pd.read_csv(train_dataset.path)
    X_train = train_df.drop("target", axis=1)
    y_train = train_df["target"]

    # Model selection
    model_map = {
        "random_forest": RandomForestClassifier,
        "gradient_boosting": GradientBoostingClassifier,
    }
    model_cls = model_map[model_type]
    model = model_cls(**hyperparameters)
    model.fit(X_train, y_train)

    # Save the model
    joblib.dump(model, trained_model.path)

    # Training metrics
    y_pred = model.predict(X_train)
    metrics.log_metric("train_accuracy", accuracy_score(y_train, y_pred))
    metrics.log_metric("train_f1", f1_score(y_train, y_pred, average="weighted"))
    metrics.log_metric("model_type", model_type)

    # Model metadata
    trained_model.metadata["framework"] = "scikit-learn"
    trained_model.metadata["model_type"] = model_type

Defining the Pipeline

Components are combined to define the whole pipeline.

from kfp import dsl, compiler

@dsl.pipeline(
    name="ml-training-pipeline",
    description="End-to-end ML training pipeline with evaluation and deployment",
)
def ml_training_pipeline(
    raw_data_path: str = "gs://my-bucket/data/raw.csv",
    test_size: float = 0.2,
    model_type: str = "random_forest",
    accuracy_threshold: float = 0.85,
):
    # Step 1: data preprocessing
    preprocess_task = preprocess_data(
        raw_data_path=raw_data_path,
        test_size=test_size,
    )
    preprocess_task.set_cpu_limit("2")
    preprocess_task.set_memory_limit("4Gi")

    # Step 2: model training
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        model_type=model_type,
        hyperparameters={
            "n_estimators": 200,
            "max_depth": 10,
            "min_samples_split": 5,
        },
    )
    train_task.set_cpu_limit("4")
    train_task.set_memory_limit("8Gi")
    train_task.set_accelerator_type("nvidia.com/gpu")
    train_task.set_accelerator_limit(1)

    # Step 3: model evaluation
    eval_task = evaluate_model(
        test_dataset=preprocess_task.outputs["test_dataset"],
        trained_model=train_task.outputs["trained_model"],
        accuracy_threshold=accuracy_threshold,
    )

    # Step 4: conditional deployment (when the accuracy threshold is exceeded)
    with dsl.Condition(
        eval_task.outputs["deploy_decision"] == "approved",
        name="check-accuracy",
    ):
        deploy_task = deploy_model(
            model=train_task.outputs["trained_model"],
            serving_endpoint="ml-model-serving",
        )

# Compile the pipeline
compiler.Compiler().compile(
    pipeline_func=ml_training_pipeline,
    package_path="ml_pipeline.yaml",
)

Advanced Pipeline Patterns

Parallel Execution and Conditional Branching

@dsl.pipeline(name="parallel-training-pipeline")
def parallel_training_pipeline(
    raw_data_path: str,
    accuracy_threshold: float = 0.85,
):
    # Data preprocessing (shared)
    preprocess_task = preprocess_data(
        raw_data_path=raw_data_path,
        test_size=0.2,
    )

    # Train several models in parallel
    models = ["random_forest", "gradient_boosting", "xgboost"]
    train_tasks = []

    for model_type in models:
        train_task = train_model(
            train_dataset=preprocess_task.outputs["train_dataset"],
            model_type=model_type,
            hyperparameters={"n_estimators": 200, "max_depth": 10},
        )
        train_task.set_display_name(f"Train {model_type}")
        train_tasks.append(train_task)

    # Select the best model
    select_task = select_best_model(
        models=[t.outputs["trained_model"] for t in train_tasks],
        metrics=[t.outputs["metrics"] for t in train_tasks],
    )

    # Deploy the champion model
    with dsl.Condition(
        select_task.outputs["best_accuracy"] >= accuracy_threshold,
        name="accuracy-gate",
    ):
        deploy_model(
            model=select_task.outputs["best_model"],
            serving_endpoint="champion-model",
        )

Recurring Runs and the Exit Handler

@dsl.pipeline(name="robust-ml-pipeline")
def robust_ml_pipeline(raw_data_path: str):
    # Exit Handler: send a notification when the pipeline completes or fails
    notify_task = send_notification(
        pipeline_name="robust-ml-pipeline",
        notification_channel="slack",
    )

    with dsl.ExitHandler(exit_task=notify_task):
        # Main pipeline logic
        preprocess_task = preprocess_data(
            raw_data_path=raw_data_path,
            test_size=0.2,
        )

        train_task = train_model(
            train_dataset=preprocess_task.outputs["train_dataset"],
            model_type="gradient_boosting",
            hyperparameters={"n_estimators": 300, "max_depth": 12},
        )

        eval_task = evaluate_model(
            test_dataset=preprocess_task.outputs["test_dataset"],
            trained_model=train_task.outputs["trained_model"],
            accuracy_threshold=0.85,
        )

# Recurring Run configuration (KFP client)
from kfp.client import Client

client = Client(host="https://kubeflow.example.com/pipeline")

# Run the pipeline every day at 2 AM
client.create_recurring_run(
    experiment_id="daily-training-exp",
    job_name="daily-model-retraining",
    pipeline_id="robust-ml-pipeline-v2",
    cron_expression="0 2 * * *",
    max_concurrency=1,
    params={
        "raw_data_path": "gs://my-bucket/data/daily/latest.csv",
    },
)

Pipeline Caching and Artifact Management

Caching Strategy

KFP supports caching: when a component's inputs are identical, the result of a previous run is reused.

# Caching configuration
@dsl.pipeline(name="cached-pipeline")
def cached_pipeline(raw_data_path: str):
    # Enable caching (default: True)
    preprocess_task = preprocess_data(
        raw_data_path=raw_data_path,
        test_size=0.2,
    )
    preprocess_task.set_caching_options(enable_caching=True)

    # Disable caching on the training step (always retrain on the latest data)
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        model_type="random_forest",
        hyperparameters={"n_estimators": 200},
    )
    train_task.set_caching_options(enable_caching=False)

Artifact Types and Management

from kfp.dsl import (
    Input, Output,
    Dataset, Model, Metrics,
    ClassificationMetrics, SlicedClassifications,
    Artifact, HTML, Markdown,
)

@dsl.component(base_image="python:3.11-slim")
def generate_evaluation_report(
    test_dataset: Input[Dataset],
    trained_model: Input[Model],
    classification_metrics: Output[ClassificationMetrics],
    html_report: Output[HTML],
    eval_metrics: Output[Metrics],
):
    """Evaluation report generation component"""
    import json

    # ClassificationMetrics: confusion matrix visualization
    classification_metrics.log_confusion_matrix(
        categories=["negative", "positive"],
        matrix=[[850, 50], [30, 270]],
    )

    # Log the ROC curve
    classification_metrics.log_roc_curve(
        fpr=[0.0, 0.1, 0.2, 0.5, 1.0],
        tpr=[0.0, 0.6, 0.8, 0.95, 1.0],
        threshold=[1.0, 0.8, 0.5, 0.2, 0.0],
    )

    # Generate the HTML report
    report_content = "<h1>Model Evaluation Report</h1>"
    report_content += "<p>Accuracy: 0.933</p>"
    report_content += "<p>F1 Score: 0.891</p>"

    with open(html_report.path, "w") as f:
        f.write(report_content)

    # Numeric metrics
    eval_metrics.log_metric("accuracy", 0.933)
    eval_metrics.log_metric("f1_score", 0.891)
    eval_metrics.log_metric("precision", 0.844)
    eval_metrics.log_metric("recall", 0.900)

Comparing Workflow Orchestration Tools

There are several tools you can use for ML workflow orchestration. Choose the right one according to the project's requirements.

Criteria for Choosing

• Kubeflow Pipelines: when you have Kubernetes infrastructure and need an ML-specific pipeline
• Airflow: when you manage data engineering and ML together and need a mature ecosystem
• Argo Workflows: when you need a Kubernetes-native general workflow engine and can build the ML-specific parts yourself
• Prefect: when you need a fast start and a flexible deployment environment

A Practical Multi-Step ML Pipeline Example

The Full Pipeline: From Data Preparation to Deployment

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==2.1.4", "great-expectations==0.18.8"],
)
def validate_data(
    raw_data_path: str,
    validated_data: Output[Dataset],
    validation_metrics: Output[Metrics],
) -> str:
    """Data quality validation"""
    import pandas as pd

    df = pd.read_csv(raw_data_path)

    # Basic data quality checks
    checks = {
        "row_count_check": len(df) > 100,
        "null_ratio_check": df.isnull().mean().max() < 0.3,
        "duplicate_check": df.duplicated().mean() < 0.05,
        "target_balance_check": df["target"].value_counts(normalize=True).min() > 0.1,
    }

    all_passed = all(checks.values())

    for check_name, passed in checks.items():
        validation_metrics.log_metric(check_name, int(passed))

    validation_metrics.log_metric("total_rows", len(df))
    validation_metrics.log_metric("all_checks_passed", int(all_passed))

    if all_passed:
        df.to_csv(validated_data.path, index=False)
        return "passed"
    else:
        failed = [k for k, v in checks.items() if not v]
        raise ValueError(f"Data validation failed: {failed}")


@dsl.component(
    base_image="python:3.11-slim",
    packages_to_install=["pandas==2.1.4", "scikit-learn==1.4.0"],
)
def feature_engineering(
    validated_data: Input[Dataset],
    feature_config: dict,
    features_dataset: Output[Dataset],
    feature_metrics: Output[Metrics],
):
    """Feature engineering"""
    import pandas as pd
    import numpy as np
    from sklearn.preprocessing import StandardScaler, LabelEncoder

    df = pd.read_csv(validated_data.path)

    # Scale the numeric features
    numeric_cols = df.select_dtypes(include=[np.number]).columns.tolist()
    numeric_cols = [c for c in numeric_cols if c != "target"]

    scaler = StandardScaler()
    df[numeric_cols] = scaler.fit_transform(df[numeric_cols])

    # Encode the categorical features
    cat_cols = df.select_dtypes(include=["object"]).columns.tolist()
    for col in cat_cols:
        le = LabelEncoder()
        df[col] = le.fit_transform(df[col].astype(str))

    df.to_csv(features_dataset.path, index=False)

    feature_metrics.log_metric("numeric_features", len(numeric_cols))
    feature_metrics.log_metric("categorical_features", len(cat_cols))
    feature_metrics.log_metric("total_features", len(df.columns) - 1)


@dsl.component(
    base_image="gcr.io/my-project/ml-serving:v1.0",
)
def deploy_to_kserve(
    model: Input[Model],
    serving_endpoint: str,
    namespace: str,
) -> str:
    """Deploy the model to KServe"""
    import subprocess
    import json
    import yaml

    inference_service = {
        "apiVersion": "serving.kserve.io/v1beta1",
        "kind": "InferenceService",
        "metadata": {
            "name": serving_endpoint,
            "namespace": namespace,
        },
        "spec": {
            "predictor": {
                "model": {
                    "modelFormat": {"name": "sklearn"},
                    "storageUri": model.uri,
                    "resources": {
                        "requests": {"cpu": "1", "memory": "2Gi"},
                        "limits": {"cpu": "2", "memory": "4Gi"},
                    },
                }
            }
        },
    }

    manifest_path = "/tmp/isvc.yaml"
    with open(manifest_path, "w") as f:
        yaml.dump(inference_service, f)

    result = subprocess.run(
        ["kubectl", "apply", "-f", manifest_path],
        capture_output=True, text=True,
    )

    if result.returncode != 0:
        raise RuntimeError(f"Deploy failed: {result.stderr}")

    return f"Deployed to {namespace}/{serving_endpoint}"


@dsl.pipeline(
    name="e2e-ml-pipeline",
    description="The full ML pipeline, from data validation to model deployment",
)
def e2e_ml_pipeline(
    raw_data_path: str = "gs://ml-data/raw/dataset.csv",
    model_type: str = "gradient_boosting",
    accuracy_threshold: float = 0.85,
    serving_endpoint: str = "fraud-detector",
    namespace: str = "ml-serving",
):
    # Exit Handler for notifications
    notify = send_notification(
        pipeline_name="e2e-ml-pipeline",
        notification_channel="slack",
    )

    with dsl.ExitHandler(exit_task=notify):
        # 1. Data validation
        validate_task = validate_data(raw_data_path=raw_data_path)

        # 2. Feature engineering
        feature_task = feature_engineering(
            validated_data=validate_task.outputs["validated_data"],
            feature_config={"scaling": "standard", "encoding": "label"},
        )

        # 3. Data split
        split_task = preprocess_data(
            raw_data_path=feature_task.outputs["features_dataset"].uri,
            test_size=0.2,
        )

        # 4. Model training
        train_task = train_model(
            train_dataset=split_task.outputs["train_dataset"],
            model_type=model_type,
            hyperparameters={"n_estimators": 300, "max_depth": 12},
        )
        train_task.set_cpu_limit("4")
        train_task.set_memory_limit("16Gi")

        # 5. Model evaluation
        eval_task = evaluate_model(
            test_dataset=split_task.outputs["test_dataset"],
            trained_model=train_task.outputs["trained_model"],
            accuracy_threshold=accuracy_threshold,
        )

        # 6. Conditional deployment
        with dsl.Condition(
            eval_task.outputs["deploy_decision"] == "approved",
            name="deploy-gate",
        ):
            deploy_to_kserve(
                model=train_task.outputs["trained_model"],
                serving_endpoint=serving_endpoint,
                namespace=namespace,
            )

# Compile and submit
compiler.Compiler().compile(
    pipeline_func=e2e_ml_pipeline,
    package_path="e2e_ml_pipeline.yaml",
)

Managing Kubernetes Resources

Pod Resources and Node Affinity

@dsl.pipeline(name="resource-managed-pipeline")
def resource_managed_pipeline():
    train_task = train_model(
        train_dataset=preprocess_task.outputs["train_dataset"],
        model_type="xgboost",
        hyperparameters={"n_estimators": 500},
    )

    # Resource limits
    train_task.set_cpu_limit("8")
    train_task.set_memory_limit("32Gi")
    train_task.set_accelerator_type("nvidia.com/gpu")
    train_task.set_accelerator_limit(2)

    # Node selector (run on GPU nodes)
    train_task.add_node_selector_constraint(
        label_name="cloud.google.com/gke-accelerator",
        value="nvidia-tesla-v100",
    )

    # Toleration settings
    train_task.set_gpu_limit(2).add_toleration(
        key="nvidia.com/gpu",
        operator="Exists",
        effect="NoSchedule",
    )

    # Mount a PVC (for large data)
    train_task.add_pvolumes({
        "/mnt/data": dsl.PipelineVolume(
            pvc="ml-data-pvc",
            volume_name="data-volume",
        ),
    })

    # Timeout setting (in seconds)
    train_task.set_timeout(3600)  # 1 hour

    # Retry settings
    train_task.set_retry(
        num_retries=3,
        policy="Always",
        backoff_duration="30s",
        backoff_factor=2.0,
        backoff_max_duration="600s",
    )

Operational Considerations

Resource Cautions

1. Memory OOM: components processing large datasets need enough memory allocated. Pandas' read_csv consumes 3-5 times the size of the data in memory.
2. GPU resource contention: when several pipelines request GPUs at the same time, Pending states drag on. Set up a ResourceQuota and a PriorityClass.
3. Concurrent PVC access: a ReadWriteOnce PVC can only be mounted by one Pod. Parallel components accessing the same PVC will fail.

Security Cautions

1. Secret management: do not pass API keys or passwords directly as pipeline parameters. Mount a Kubernetes Secret as an environment variable.
2. Image vulnerabilities: scan base images for security vulnerabilities regularly. Consider a distroless image instead of python:3.11-slim.
3. RBAC configuration: apply the principle of least privilege to the pipeline service account.

Failure Cases and Recovery Procedures

Case 1: Pod OOMKilled

Symptom: the component Pod fails in the OOMKilled state

# Check the Pod status
kubectl get pods -n kubeflow -l pipeline/runid=run-abc123
kubectl describe pod train-model-xxxxx -n kubeflow

# Look for OOMKilled in the events
# Last State: Terminated
#   Reason: OOMKilled
#   Exit Code: 137

Recovery procedure:

# 1. Raise the memory limit
train_task.set_memory_limit("64Gi")

# 2. Modify the component to process the data in chunks
@dsl.component(base_image="python:3.11-slim")
def train_with_chunks(
    train_dataset: Input[Dataset],
    chunk_size: int,
    trained_model: Output[Model],
):
    import pandas as pd
    from sklearn.linear_model import SGDClassifier

    model = SGDClassifier(loss="log_loss")
    chunks = pd.read_csv(train_dataset.path, chunksize=chunk_size)

    for chunk in chunks:
        X = chunk.drop("target", axis=1)
        y = chunk["target"]
        model.partial_fit(X, y, classes=[0, 1])

    import joblib
    joblib.dump(model, trained_model.path)

Case 2: Pipeline Version Conflict

Symptom: an existing Recurring Run fails after a pipeline update

Recovery procedure:

from kfp.client import Client

client = Client(host="https://kubeflow.example.com/pipeline")

# 1. Disable the existing Recurring Run
client.disable_recurring_run(recurring_run_id="run-xxx")

# 2. Upload the new pipeline version
pipeline_version = client.upload_pipeline_version(
    pipeline_package_path="ml_pipeline_v3.yaml",
    pipeline_version_name="v3.0",
    pipeline_id="ml-training-pipeline",
)

# 3. Create the new Recurring Run
client.create_recurring_run(
    experiment_id="daily-training-exp",
    job_name="daily-model-retraining-v3",
    version_id=pipeline_version.pipeline_version_id,
    cron_expression="0 2 * * *",
    max_concurrency=1,
)

Case 3: Metadata DB Connection Failure

Symptom: artifact tracking fails because of an ML Metadata Service connection error

# Check the MLMD service status
kubectl get pods -n kubeflow -l app=metadata-grpc-server
kubectl logs metadata-grpc-server-xxxxx -n kubeflow

# Check the MySQL/PostgreSQL connection
kubectl exec -it metadata-grpc-server-xxxxx -n kubeflow -- \
    mysql -h metadata-db -u root -p -e "SHOW DATABASES;"

# Restart the MLMD service
kubectl rollout restart deployment metadata-grpc-server -n kubeflow

Production Checklist

Infrastructure Setup

• Kubeflow Pipelines installed on the Kubernetes cluster and the version confirmed (KFP v2 recommended)
• Artifact Store (MinIO/S3/GCS) configured and access permissions confirmed
• Metadata DB (MySQL/PostgreSQL) configured for high availability
• RBAC and namespace isolation configured
• GPU node pool and autoscaling configured

Pipeline Development

• Resource limits (CPU/Memory/GPU) set on every component
• Retry policy and timeouts set
• Caching strategy decided (which steps to cache)
• Default values set for pipeline parameters
• A data validation component included

Operations and Monitoring

• Recurring Runs configured with a concurrency limit
• Pipeline failure notifications (Slack/PagerDuty) configured
• Artifact storage capacity monitored
• MLMD backup schedule configured
• A retention policy for pipeline run history

Security

• Container image vulnerability scanning automated
• Sensitive information managed as Kubernetes Secrets
• Pod-to-Pod communication restricted with network policies
• Principle of least privilege applied to service accounts

References

• Kubeflow Pipelines Official Documentation
• KFP v2 SDK Reference
• Kubeflow Pipelines GitHub Repository
• KServe Official Documentation - Model Serving
• ML Metadata (MLMD) Official Documentation
• Argo Workflows Official Documentation
• Google Cloud Vertex AI Pipelines Guide