Demo Pipeline Control Flow

This lesson demonstrates how to execute a component (task) multiple times in parallel inside a Kubeflow Pipeline, collect outputs from each parallel run, and then select the best result. Use case summary:

Run a train_and_evaluate component for several learning_rate values in parallel.
Aggregate accuracies and learning rates from each parallel iteration.
Use a selector component (pick_best_learning_rate) to choose the learning rate that produced the highest accuracy.

Key ideas used:

ParallelFor to iterate and launch parallel runs.
Collected to gather outputs from parallel iterations into Python lists.
NamedTuple return type so component outputs are named and accessible by downstream steps.

Quick overview of tools in this example:

Resource / Construct	Purpose	Example usage
Component	Encapsulated, reusable pipeline step	`@component def train_and_evaluate(...)`
`ParallelFor`	Launch the same component across multiple inputs in parallel	`with ParallelFor(learning_rates) as lr:`
`Collected`	Aggregate one output across all parallel iterations into a list	`Collected(run.outputs["accuracy"])`
Named outputs	Make outputs addressable by name in `run.outputs`	`-> NamedTuple("Outputs", [("accuracy", float), ("learning_rate", float)])`

Below is a compact, corrected, and working pipeline example that demonstrates this pattern.

from kfp.dsl import component, pipeline, ParallelFor, Collected
from kfp import compiler
from typing import NamedTuple, List
import random

@component
def train_and_evaluate(learning_rate: float) -> NamedTuple(
    "Outputs",
    [
        ("accuracy", float),
        ("learning_rate", float),
    ],
):
    """
    Dummy training component that returns a random accuracy and echoes the learning rate.
    """
    # Simulate training and evaluation
    accuracy = random.uniform(0.90, 1.00)
    return accuracy, learning_rate

@component
def pick_best_learning_rate(accuracies: List[float], learning_rates: List[float]) -> float:
    """
    Selects and returns the learning rate that produced the best (maximum) accuracy.
    """
    # Pair learning rates with accuracies and find the pair with the max accuracy
    pairs = list(zip(learning_rates, accuracies))
    best_pair = max(pairs, key=lambda x: x[1])

    print(f"Best LR = {best_pair[0]} with accuracy = {best_pair[1]}")
    # Return the best learning rate as a float so downstream components can consume it.
    return float(best_pair[0])

@pipeline(name="lr-sweep-metrics-only")
def ml_pipeline():
    # The list of learning rates to sweep over
    learning_rates = [0.001, 0.01, 0.1]

    # Run train_and_evaluate in parallel for each value in learning_rates
    with ParallelFor(learning_rates) as lr:
        run = train_and_evaluate(learning_rate=lr)

    # Collect outputs from the parallel runs into Python lists
    accuracies = Collected(run.outputs["accuracy"])
    lrs_collected = Collected(run.outputs["learning_rate"])

    # Pass collected lists to a component that picks the best learning rate
    best_lr = pick_best_learning_rate(
        accuracies=accuracies,
        learning_rates=lrs_collected,
    )

if __name__ == "__main__":
    compiler.Compiler().compile(pipeline_func=ml_pipeline, package_path="par.yaml")

How it works — step-by-step:

Define train_and_evaluate to return a NamedTuple with ("accuracy", float) and ("learning_rate", float).
In the pipeline, prepare the list of learning rates to try: [0.001, 0.01, 0.1].
Use with ParallelFor(learning_rates) as lr: to execute train_and_evaluate(learning_rate=lr) for each value concurrently.
Use Collected(run.outputs["accuracy"]) and Collected(run.outputs["learning_rate"]) to aggregate each output across all parallel iterations into lists.
Call pick_best_learning_rate with the collected lists; it zips them, finds the maximum accuracy, and returns the corresponding learning rate.
Compile the pipeline to a package (here par.yaml) and upload it to Kubeflow Pipelines.

When executed in the Kubeflow UI you will see a DAG representing each parallel iteration for each learning rate; each iteration runs train-and-evaluate with the corresponding learning_rate. After all iterations finish, the pick-best-learning-rate step runs using the collected lists.

A screenshot of the Kubeflow Central Dashboard showing a pipeline run titled "Run of parallel (90bee)" with a selected "train-and-evaluate" step displaying an input parameter "learning_rate" set to 0.001. The left sidebar shows navigation items like Home, Notebooks, TensorBoards, and Pipelines.

Example viewable outputs after a run (example random outputs from this dummy training):

[
  0.9236584810034232,
  0.9306029262789163,
  0.9145041373531706
]

[
  0.001,
  0.01,
  0.1
]

Output
0.01

Use Collected only on an output of a ParallelFor iteration (for example, run.outputs["accuracy"]). Collected aggregates that output over all iterations into a list that can be passed to downstream components.

This pattern is ideal for hyperparameter sweeps, ensemble experiments, or any workflow that runs the same step with different inputs and then aggregates and selects the best result. Links and references:

Kubeflow Pipelines: https://www.kubeflow.org/docs/components/pipelines/
KFP SDK: https://github.com/kubeflow/pipelines
Example concepts: ParallelFor, Collected, component outputs (NamedTuple)

Course Introduction

Fundamentals of Kubeflow

Working With Kubeflow

KServe and Katib

Profiles and Multi-Tenancy

Bringing It All Together

Demo Pipeline Control Flow

Watch Video