Fendlemark v0.9.12

How-to

Tracking Kubeflow Pipelines runs in MLflow

By the Fendlemark Keepers ·

A training job that Kubeflow Pipelines runs and MLflow records has three ids: the pipeline run's, the Argo Workflow's underneath it, and the MLflow run's beside it. Nothing joins them unless you write the join. This guide writes it, with the Kubeflow Pipelines (KFP) v2 SDK and a few lines of MLflow, and starts by checking how much of it your installation already does. None of it needs Fendlemark; the last section shows what the Fendlemark app does with the result.

By the end, each MLflow run that a pipeline task opens or resumes will carry the pipeline run's id and display name as tags. The training task will hand its MLflow run id to later tasks, and evaluation will log into the training run instead of starting another. The examples use a pipeline named fraud-score in the Kubeflow profile team-risk, and an MLflow tracking server at http://mlflow.mlflow.svc:5000. Substitute your own.

1. Check whether KFP's MLflow integration is on

Since release 2.17, Kubeflow Pipelines has an MLflow integration. An administrator turns it on for the whole installation, in the API server's configuration. Each namespace can then change a few of its settings, such as the default experiment and whether tasks get the MLflow variables, and a single run can opt out through the API.

Where it is on, each pipeline run is recorded as a parent MLflow run, named after the pipeline run and tagged kfp.pipeline_run_id, in an experiment named KFP-Default unless someone chooses another. Each task gets a nested MLflow run, named after the task. Run parameters and scalar metric artifacts are logged for you, and when the pipeline run ends, Kubeflow Pipelines closes any nested run still open. If injection of the MLflow variables is also on (it is off by default), each task starts with MLFLOW_TRACKING_URI, MLFLOW_EXPERIMENT_ID, and MLFLOW_RUN_ID set, the last naming the task's own nested run.

To tell whether the integration is on, look in MLflow for a run tagged kfp.pipeline_run_id with a pipeline run's id. To tell whether tasks get the variables, print os.environ.get("MLFLOW_RUN_ID") from a task. That gives three cases, and the rest of this guide handles all of them:

2. Give tasks the tracking server

The MLflow client reads its tracking server from MLFLOW_TRACKING_URI, and credentials for MLflow's basic authentication from MLFLOW_TRACKING_USERNAME and MLFLOW_TRACKING_PASSWORD. Keep them in a ConfigMap and a Secret in the profile's namespace, where the pipeline's pods run:

$ kubectl -n team-risk create configmap mlflow-client \
    --from-literal=MLFLOW_TRACKING_URI=http://mlflow.mlflow.svc:5000
$ kubectl -n team-risk create secret generic mlflow-credentials \
    --from-literal=MLFLOW_TRACKING_USERNAME=kfp-team-risk \
    --from-literal=MLFLOW_TRACKING_PASSWORD="$MLFLOW_PASSWORD"

The kfp-kubernetes package (pip install kfp-kubernetes) attaches both to a task as environment variables. A small helper keeps the pipeline readable:

from kfp import kubernetes


def with_mlflow(task):
    """Give a task the tracking server's address and credentials."""
    kubernetes.use_config_map_as_env(
        task,
        config_map_name="mlflow-client",
        config_map_key_to_env={"MLFLOW_TRACKING_URI": "MLFLOW_TRACKING_URI"},
    )
    kubernetes.use_secret_as_env(
        task,
        secret_name="mlflow-credentials",
        secret_key_to_env={
            "MLFLOW_TRACKING_USERNAME": "MLFLOW_TRACKING_USERNAME",
            "MLFLOW_TRACKING_PASSWORD": "MLFLOW_TRACKING_PASSWORD",
        },
    )
    return task

Moving the tracking server later is then one ConfigMap edit, with no pipeline to recompile.

3. Give each task the pipeline run's identity

A task cannot ask Kubeflow Pipelines which run it belongs to, but the SDK has placeholders that the backend fills in at run time. Pass them to a component as ordinary string arguments. According to the SDK reference, four of them identify where a task is running:

PlaceholderValue
dsl.PIPELINE_JOB_ID_PLACEHOLDERThe pipeline run's UUID: the run id in the API, the UI's URLs, and kfp.Client().get_run().
dsl.PIPELINE_JOB_NAME_PLACEHOLDERThe pipeline run's display name, given when the run was started.
dsl.PIPELINE_JOB_RESOURCE_NAME_PLACEHOLDERThe run's name in the underlying engine, which on the default backend is the Argo Workflow's name.
dsl.PIPELINE_TASK_NAME_PLACEHOLDERThe task's name in the pipeline, which defaults to its component's name.

The run's UUID is the one to keep. It is unique, the API looks runs up by it, and Kubeflow Pipelines also sets it as the pipeline/runid label on the run's Argo Workflow, so one value finds both the pipeline run and its Workflow. The display name is for people, and it repeats.

Here is a three-task pipeline that loads features, trains, and evaluates, using with_mlflow from step 2. The training task receives the pipeline run's id and display name, and returns the MLflow run id as its output:

from kfp import compiler, dsl


@dsl.component(base_image="python:3.12", packages_to_install=["pandas"])
def load_features(features: dsl.Output[dsl.Dataset]):
    ...  # writes the training set to features.path


@dsl.component(base_image="python:3.12", packages_to_install=["mlflow", "pandas", "scikit-learn"])
def train(
    features: dsl.Input[dsl.Dataset],
    n_estimators: int,
    kfp_run_id: str,
    kfp_run_name: str,
    model: dsl.Output[dsl.Model],
) -> str:
    import os

    import mlflow

    tags = {"kfp.run_id": kfp_run_id, "kfp.run_name": kfp_run_name}
    if "MLFLOW_RUN_ID" in os.environ:
        # KFP's MLflow integration opened this task's run: resume it, and keep its name.
        run = mlflow.start_run(tags=tags)
    else:
        mlflow.set_experiment("fraud-score")
        run = mlflow.start_run(run_name=kfp_run_name, tags=tags)
    with run:
        mlflow.log_param("n_estimators", n_estimators)
        ...  # trains, logs metrics with mlflow.log_metrics, writes the model to model.path
        return run.info.run_id


@dsl.component(base_image="python:3.12", packages_to_install=["mlflow", "scikit-learn"])
def evaluate(mlflow_run_id: str, model: dsl.Input[dsl.Model]):
    import mlflow

    # Log into the training run rather than a run of this task's own.
    with mlflow.start_run(run_id=mlflow_run_id):
        ...  # evaluates the model and logs the results with mlflow.log_metrics


@dsl.pipeline(name="fraud-score")
def fraud_score(n_estimators: int = 400):
    features = load_features()
    trained = with_mlflow(train(
        features=features.outputs["features"],
        n_estimators=n_estimators,
        kfp_run_id=dsl.PIPELINE_JOB_ID_PLACEHOLDER,
        kfp_run_name=dsl.PIPELINE_JOB_NAME_PLACEHOLDER,
    ))
    trained.set_caching_options(enable_caching=False)
    evaluated = with_mlflow(evaluate(mlflow_run_id=trained.outputs["Output"], model=trained.outputs["model"]))
    evaluated.set_caching_options(enable_caching=False)


compiler.Compiler().compile(fraud_score, package_path="fraud-score.yaml")

A few choices in there are deliberate:

Start a run from the compiled fraud-score.yaml as you would for any pipeline: upload it in the UI, or pass it to kfp.Client().create_run_from_pipeline_package().

4. Find one from the other

From a pipeline run to its MLflow runs. Copy the run id from the run's page in the Kubeflow Pipelines UI and search MLflow by tag:

import mlflow

runs = mlflow.search_runs(
    filter_string="tags.`kfp.run_id` = '8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34'",
    search_all_experiments=True,
)
print(runs[["run_id", "status", "tags.kfp.run_name"]])

The backticks around the tag name are optional here, since MLflow splits a tag filter at its first dot, so tags.kfp.run_id works as well; a name with hyphens or spaces needs them. The value goes in single or double quotes. The same filter works in the MLflow UI's search box. Searching every experiment matters: with the integration on, the run is in whichever experiment the integration chose, not necessarily fraud-score.

From an MLflow run to its pipeline run. The kfp.run_id tag is everything the Kubeflow Pipelines client needs:

from kfp.client import Client

run = Client().get_run(run_id="8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34")
print(run.display_name, run.state)

In Kubeflow's Central Dashboard, the same run is at /pipeline/#/runs/details/8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34.

From either to the Argo Workflow. Kubeflow Pipelines labels every Workflow it creates with the run id, so the same id finds the Workflow:

$ kubectl -n team-risk get workflows -l pipeline/runid=8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34
NAME                STATUS      AGE   MESSAGE
fraud-score-q8w4z   Succeeded   41m

Things that will happen eventually

How Fendlemark correlates them

If you run Fendlemark with the kubeflow, argo, and mlflow adapters enabled, with team-risk among the kubeflow adapter's profiles and the argo adapter's namespaces, the tag from step 3 is all it needs. Running the Fendlemark app on Kubernetes sets that up.

The kubeflow adapter gives every pipeline run facet the correlation key kfp.run_id. The argo adapter reads the pipeline/runid label into the same key, so a pipeline run and its Workflow merge without any tagging; the correlation matrix marks that cell "automatic". The mlflow adapter reads the run tags listed in correlation_tags, whose default list includes kfp.run_id. The lattice sorts the facets that carry the same value by start time, and links each to the next when they come from different adapters and the gap between them is within lattice.correlation.window:

$ fendlemark lattice inspect --key kfp.run_id=8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34
ID                   KIND  STATUS     FACETS                STARTED               TITLE
lt_47271ae5a6095e8e  run   Succeeded  argo+kubeflow+mlflow  2026-10-28T14:00:00Z  fraud-score 2026-10-28
$ fendlemark lattice explain lt_47271ae5a6095e8e
lt_47271ae5a6095e8e  run  Succeeded  "fraud-score 2026-10-28"
  facet  mlflow     2a7d9f4c1e8b4c6d9a3f5e7b1c0d2e48
  facet  argo       team-risk/fraud-score-q8w4z
  facet  kubeflow   run:8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34
  merge  kfp.run_id = "8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34" linked kubeflow/run:8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34 and argo/team-risk/fraud-score-q8w4z (gap 3s)
  merge  kfp.run_id = "8c2f4e1d-3b7a-4f90-b6d2-7e1a9c5f0d34" linked argo/team-risk/fraud-score-q8w4z and mlflow/2a7d9f4c1e8b4c6d9a3f5e7b1c0d2e48 (gap 1m9s)

Both merges come from one key: the first from the label Kubeflow Pipelines set, the second from the tag your code set. The entity is a run, because the MLflow facet outranks the two workflow facets. It also takes the MLflow run's name as its title, so naming a new MLflow run after the pipeline run names the entity too; with the integration's variables, the title is the task's name, train. In the Fendlemark console the entity appears once, in the Runs view, with a panel for each tool and a deep link into each (the Kubeflow link needs dashboard_url, and the MLflow link a tracking_uri that browsers can open).

The second gap is the one to watch. The Workflow starts a few seconds after the pipeline run is created, and the MLflow run only when the training task does, after load_features has finished: here 69 seconds after the Workflow, inside the default window of 90 seconds. A feature load that takes ten minutes would leave the MLflow run as an entity of its own. There are two fixes. Widen the window to cover the tasks before training:

lattice:
  correlation:
    window: 20m

Or open the MLflow run in the first task and pass its id along, as step 3 already does from training onwards. That also settles retries.

Two facets from the same adapter never merge. Of several MLflow runs with the same tag, such as retried attempts or loop iterations, the entity takes the first to start, and the rest stay entities of their own. For a retried task the first is the failed attempt, so the entity shows Failed even though the pipeline run succeeded, which is one more reason to open the run in a task that does not retry. fendlemark lattice inspect --key kfp.run_id=… lists them all, and lattice explain shows the gap of every merge.

Fendlemark does not read the kfp.pipeline_run_id tag on the integration's parent run. Adding it to correlation_tags would only produce a key of that name, which no kubeflow facet carries, so the kfp.run_id tag from step 3 is what joins the runs, with or without the integration.

The join also works the other way round. Open the MLflow run before the pipeline starts, pass its id in as a pipeline parameter named mlflow_run_id, and have the tasks resume it with mlflow.start_run(run_id=…) instead of opening their own. The kubeflow adapter reads the parameter as the key mlflow.run_id, so the MLflow run needs no tag at all, provided the pipeline run is created within lattice.correlation.window of the MLflow run's start.

Further reading