Experiments & Run Tracking

Run experiments, compare pipeline runs, track metrics and artifacts, and implement systematic hyperparameter tuning with KubeFlow Pipelines.

Creating Experiments

import kfp

client = kfp.Client(host="http://localhost:8080")

# Create an experiment
experiment = client.create_experiment(
    name="model-optimization-v2",
    description="Testing different model architectures and hyperparameters"
)

# Submit a pipeline run to the experiment
run = client.create_run_from_pipeline_func(
    training_pipeline,
    experiment_name="model-optimization-v2",
    arguments={"data_url": "gs://bucket/data.csv", "lr": 0.001},
    run_name="lr-0.001-run"
)

Logging Metrics

from kfp import dsl
from kfp.dsl import Metrics, Output

@dsl.component
def evaluate_model(model: dsl.Input[dsl.Model], test_data: str,
                   metrics: Output[Metrics]) -> float:
    # ... load model and evaluate ...
    accuracy = 0.95
    precision = 0.93
    recall = 0.91

    metrics.log_metric("accuracy", accuracy)
    metrics.log_metric("precision", precision)
    metrics.log_metric("recall", recall)
    metrics.log_metric("f1_score", 2 * precision * recall / (precision + recall))
    return accuracy

Comparing Runs

The KFP UI provides built-in run comparison features. You can also programmatically retrieve metrics:

# List runs in an experiment
runs = client.list_runs(experiment_id=experiment.experiment_id)

# Compare metrics across runs
for run in runs.runs:
    metrics = client.get_run(run.run_id)
    print(f"Run: {run.display_name}, Status: {run.status}")

Recurring Runs

# Create a recurring run (e.g., daily retraining)
recurring_run = client.create_recurring_run(
    experiment_id=experiment.experiment_id,
    job_name="daily-retraining",
    pipeline_id=pipeline.pipeline_id,
    cron_expression="0 2 * * *",  # Every day at 2 AM
    params={"data_url": "gs://bucket/latest-data.csv"}
)