kubeflow-pipelines

Kubeflow Pipelines v2

Repository: github.com/kubeflow/pipelines
KFP SDK: latest ~2.16.0
API version: pipelines.kubeflow.org/v2beta1 (Kubernetes Native API)
Compile target: IR YAML (not Argo Workflow YAML)

Architecture

Python DSL (@dsl.component, @dsl.pipeline)
     │
     ▼ compile()
IR YAML (intermediate representation)
     │
     ▼ run/create
KFP Backend / Kubernetes Native API

Python SDK

Installation

pip install kfp

Components

@dsl.component (Python function)

from kfp import dsl

@dsl.component(base_image="python:3.11")
def train_model(
    dataset: dsl.Dataset,
    model: dsl.Output[dsl.Model],
    metrics: dsl.Output[dsl.Metrics],
    epochs: int = 10,
    lr: float = 0.001,
) -> str:
    """Train a model on the input dataset."""
    import json

    # Training logic here
    accuracy = 0.95

    # Output metrics
    metrics.log_metric("accuracy", accuracy)
    metrics.log_metric("loss", 0.023)

    # Model output path
    model.path = "/tmp/model.pkl"

    return f"Model trained with accuracy {accuracy}"

@dsl.container_component (pre-built container)

@dsl.container_component
def preprocess_data(
    input_data: dsl.Input[dsl.Dataset],
    output_data: dsl.Output[dsl.Dataset],
):
    return dsl.ContainerSpec(
        image="my-registry/preprocessor:latest",
        command=["./preprocess.sh"],
        args=[
            "--input", input_data.uri,
            "--output", output_data.uri,
        ],
    )

Importer (external artifact)

from kfp import dsl

@dsl.pipeline
def my_pipeline():
    # Import an existing artifact
    data = dsl.importer(
        artifact_uri="s3://bucket/datasets/mnist/",
        artifact_class=dsl.Dataset,
        reimport=False,
    )

Pipelines

@dsl.pipeline(
    name="training-pipeline",
    description="End-to-end model training pipeline",
    pipeline_root="s3://bucket/pipeline-runs/",
)
def training_pipeline(
    epochs: int = 10,
    lr: float = 0.001,
    model_name: str = "resnet50",
):
    # Component invocations
    preprocess_task = preprocess_data(
        input_data=dsl.importer(
            artifact_uri="s3://bucket/raw/",
            artifact_class=dsl.Dataset,
        ).output
    )

    train_task = train_model(
        dataset=preprocess_task.outputs["output_data"],
        epochs=epochs,
        lr=lr,
    )

    # Set resource limits per task
    train_task.set_cpu_limit("8")
    train_task.set_memory_limit("32Gi")
    train_task.set_accelerator_limit("nvidia.com/gpu", 4)
    train_task.set_caching_options(False)