Implementation:Kubeflow Kubeflow MLMD Pipelines Dashboard

Knowledge Sources	Kubeflow Pipelines README Kubeflow Central Dashboard README
Domains	MLOps, Monitoring, Metadata Tracking
Last Updated	2026-02-13 00:00 GMT

Overview

Concrete tool for monitoring ML pipeline health, querying artifact lineage, and comparing model performance provided by Kubeflow Pipelines UI, ML Metadata, and the Central Dashboard.

Description

The monitoring and iteration capability in Kubeflow is delivered through three integrated components. The Kubeflow Pipelines UI provides a web interface for viewing pipeline run history, visualizing DAG execution, inspecting individual step inputs/outputs/logs, and comparing metrics across runs. ML Metadata (MLMD) is the underlying metadata store that records all artifacts, executions, and contexts produced by pipeline runs, enabling programmatic lineage queries and artifact tracking. The Central Dashboard serves as the unified entry point for all Kubeflow components, providing namespace-scoped navigation to Notebooks, Pipelines, Training Jobs, Model Registry, and Katib Experiments from a single interface.

Together, these tools provide the observability layer for the ML lifecycle. Engineers can trace any deployed model back through its serving configuration, registry entry, training job, pipeline run, and original dataset. They can compare metrics across pipeline runs or model versions to identify improvements or regressions. And they can use MLMD queries programmatically to build automated drift detection and retraining trigger logic.

External Reference

Usage

Use the MLMD, Pipelines UI, and Central Dashboard when:

Pipeline run history must be reviewed for failures, execution times, or output inspection.
Model metrics need to be compared across pipeline runs or experiment variants.
Artifact lineage must be traced from a deployed model back to its training data and code.
Programmatic queries against metadata are needed for automated monitoring or retraining triggers.
A unified view across all Kubeflow components is needed for operational oversight.

Code Reference

Source Location

Repository: kubeflow/pipelines
Dashboard Repository: kubeflow/dashboard
MLMD Repository: google/ml-metadata
File: frontend/src/ (Pipelines UI), backend/src/apiserver/ (Pipelines API server)

Signature

# Programmatic MLMD queries via the KFP Python client
import kfp

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

# List pipeline runs
runs = client.list_runs(experiment_id="exp-123", sort_by="created_at desc")

# Get run details including metrics and artifacts
run = client.get_run(run_id="run-abc")

# Compare runs
run_a = client.get_run(run_id="run-a")
run_b = client.get_run(run_id="run-b")

# Direct MLMD queries via ml-metadata Python library
from ml_metadata import metadata_store
from ml_metadata.proto import metadata_store_pb2

connection_config = metadata_store_pb2.MetadataStoreClientConfig(
    host="metadata-grpc-service.kubeflow",
    port=8080,
)
store = metadata_store.MetadataStore(connection_config)

# Query artifacts by type
models = store.get_artifacts_by_type("system.Model")

Import

# Install KFP client for programmatic pipeline queries
pip install kfp==2.11.0

# Install ml-metadata for direct MLMD queries
pip install ml-metadata

# Access the Pipelines UI via Central Dashboard
# Navigate to: https://<kubeflow-host>/pipeline/

# Access the Central Dashboard
# Navigate to: https://<kubeflow-host>/

I/O Contract

Inputs

Name	Type	Required	Description
experiment_id	string	No	Filter runs by experiment for scoped comparison
run_id	string	No	Specific pipeline run to inspect
artifact_type	string	No	MLMD artifact type to query (e.g., system.Model, system.Dataset)
sort_by	string	No	Sort order for listing runs (e.g., created_at desc)
namespace	string	No	Kubernetes namespace to scope dashboard views
drift_detection_threshold	float	No	Threshold for automated drift detection on feature distributions

Outputs

Name	Type	Description
Pipeline run history	list of Run objects	Historical record of all pipeline executions with status and timing
Run metrics	dict	Metrics logged during pipeline execution (accuracy, loss, etc.)
Artifact lineage	MLMD lineage graph	Provenance chain linking artifacts to their producing executions
Run comparison	comparison view	Side-by-side metric and parameter comparison across selected runs
Dashboard navigation	web UI	Unified access to all Kubeflow component UIs within the namespace

Usage Examples

Basic Usage

import kfp

# Connect to the Kubeflow Pipelines backend
client = kfp.Client(host="https://kubeflow.example.com/pipeline")

# List recent pipeline runs for an experiment
runs = client.list_runs(
    experiment_id="exp-train-fraud-detector",
    sort_by="created_at desc",
    page_size=10,
)

# Print run summaries
for run in runs.runs:
    print(
        f"Run: {run.run_id}, "
        f"Status: {run.state}, "
        f"Created: {run.created_at}"
    )

# Get detailed metrics from a specific run
run_detail = client.get_run(run_id="run-latest-abc")
print(f"Run metrics: {run_detail.run.metrics}")

MLMD Lineage Query

from ml_metadata import metadata_store
from ml_metadata.proto import metadata_store_pb2

# Connect to the MLMD gRPC service
connection_config = metadata_store_pb2.MetadataStoreClientConfig(
    host="metadata-grpc-service.kubeflow.svc.cluster.local",
    port=8080,
)
store = metadata_store.MetadataStore(connection_config)

# Get all registered model artifacts
model_artifacts = store.get_artifacts_by_type("system.Model")

# For each model, trace back to its producing execution
for model in model_artifacts:
    events = store.get_events_by_artifact_ids([model.id])
    producing_executions = [
        e.execution_id for e in events
        if e.type == metadata_store_pb2.Event.OUTPUT
    ]
    print(
        f"Model: {model.uri}, "
        f"Produced by executions: {producing_executions}"
    )

# Query dataset artifacts to check for drift
datasets = store.get_artifacts_by_type("system.Dataset")
latest_dataset = sorted(datasets, key=lambda d: d.create_time_since_epoch)[-1]
print(f"Latest dataset: {latest_dataset.uri}")

Automated Retraining Trigger

import kfp

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

def check_and_retrain(
    pipeline_id: str,
    experiment_id: str,
    accuracy_threshold: float = 0.90,
):
    """Check latest model accuracy and trigger retraining if below threshold."""
    runs = client.list_runs(
        experiment_id=experiment_id,
        sort_by="created_at desc",
        page_size=1,
    )

    if runs.runs:
        latest_run = runs.runs[0]
        run_detail = client.get_run(run_id=latest_run.run_id)

        for metric in run_detail.run.metrics:
            if metric.name == "accuracy" and metric.number_value < accuracy_threshold:
                print(
                    f"Accuracy {metric.number_value} below threshold "
                    f"{accuracy_threshold}. Triggering retraining."
                )
                client.run_pipeline(
                    experiment_id=experiment_id,
                    pipeline_id=pipeline_id,
                    job_name="automated-retraining",
                )
                return

    print("Model accuracy is within acceptable range.")

check_and_retrain(
    pipeline_id="pipeline-train-fraud-detector",
    experiment_id="exp-train-fraud-detector",
    accuracy_threshold=0.92,
)

Related Pages

Implements Principle

Principle:Kubeflow_Kubeflow_Monitor_And_Iterate

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment