# Tracking Kubeflow Pipelines runs in MLflow

Source: https://fendlemark.app/blog/tracking-kubeflow-pipelines-runs-in-mlflow/

By [the Fendlemark Keepers](https://fendlemark.app/governance/#keepers) · 29 October 2026

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:

- **No integration.** Step 2 gives tasks the tracking server, and step 3's code opens one MLflow run per pipeline run.
- **The integration, with its variables.** Skip step 2 and leave out the `with_mlflow` calls, since the tasks already have the tracking server. `mlflow.start_run()` resumes the nested run the integration made for the task, and step 3 adds the tags that let you find it from the pipeline run.
- **The integration, without its variables.** The pipeline run still appears in MLflow, but the tasks know nothing about it, so the code behaves as it does with no integration and opens its own run beside the integration's. The two share the pipeline run's id, under `kfp.run_id` on yours and `kfp.pipeline_run_id` on the integration's parent.

## 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:

```shell
$ 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:

```python
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:

| Placeholder | Value |
| --- | --- |
| `dsl.PIPELINE_JOB_ID_PLACEHOLDER` | The pipeline run's UUID: the run id in the API, the UI's URLs, and `kfp.Client().get_run()`. |
| `dsl.PIPELINE_JOB_NAME_PLACEHOLDER` | The pipeline run's display name, given when the run was started. |
| `dsl.PIPELINE_JOB_RESOURCE_NAME_PLACEHOLDER` | The run's name in the underlying engine, which on the default backend is the Argo Workflow's name. |
| `dsl.PIPELINE_TASK_NAME_PLACEHOLDER` | The 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:

```python
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:

- **The tags go on the run, whoever opened it.** `mlflow.start_run()` sets its `tags` on a new run and on a resumed one alike, so the same tags land on the run this task opens, or on the nested run that the integration opened for it.
- **The experiment is chosen only for a new run.** If `MLFLOW_RUN_ID` names a run in another experiment, calling `mlflow.set_experiment()` first makes `start_run()` raise, because the active experiment no longer matches the run.
- **A new run is named after the pipeline run, and a resumed one is left alone.** MLflow does not require run names to be unique, and a name that matches the Kubeflow Pipelines UI makes the two easy to match by eye. Since MLflow 3.6, `start_run()` also renames a resumed run to any `run_name` it is given, whatever its docstring says, so the code passes none when resuming and the integration's task run keeps its name.
- **The MLflow run id is the task's output.** A component that returns a value and also has output artifacts exposes the return value as `outputs["Output"]`. The value also shows in the task's details in the Kubeflow Pipelines UI, which is the quickest way from a pipeline run to its MLflow run.
- **The tag is called `kfp.run_id`.** Any name would do for MLflow. This is the one Fendlemark reads, as the last section explains.

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:

```python
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:

```python
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:

```shell
$ 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

- **A cache hit skips your code.** When a task is a cache hit, Kubeflow Pipelines reuses the outputs of an earlier execution and does not run the container. No MLflow run opens, the id the task appears to return is the earlier execution's, and evaluation then logs into an MLflow run that belongs to a different pipeline run. In this pipeline that cannot happen: the cache key includes the resolved input values, and the run id argument differs on every run, so neither tracked task can be a cache hit. Turning caching off says so explicitly, and protects a copy of the pipeline that drops the argument. (With the integration on, a cache hit still gets an MLflow run, but none of your code ran in it.)
- **Retries make more runs, and the first one failed.** Without the integration's variables, a task with `set_retry()` opens a new MLflow run on every attempt, all with the same `kfp.run_id` tag. Evaluation logs into the successful attempt's run, because only a successful attempt produces an output. To keep one MLflow run per pipeline run, open it in an earlier task that does not retry, pass its id in, and have the retried task resume it with `mlflow.start_run(run_id=…)`.
- **Loops make more runs, on purpose.** A training task inside `dsl.ParallelFor` runs once per item, and each run carries the same `kfp.run_id`. That is usually what you want for a sweep. Without the integration, group them by opening a parent run in an earlier task, passing its id in, and giving each iteration `mlflow.start_run(parent_run_id=…, nested=True)`. With it, each iteration already has a nested run under the pipeline run's parent.
- **Killed pods leave running runs.** Without the integration, a pod killed for exceeding its memory limit never reaches the end of its `with` block, so its MLflow run stays `RUNNING` after the pipeline run has failed. [The MLflow and Argo Workflows post](https://fendlemark.app/blog/running-mlflow-and-argo-workflows-together/#gotchas) shows how to find such runs; the search is the same here. With the integration, Kubeflow Pipelines closes the task's nested run when the pipeline run ends.
- **Display names repeat.** Two pipeline runs started as "fraud-score nightly" have the same `kfp.run_name`. Search by `kfp.run_id` whenever it matters which one you mean.

## How Fendlemark correlates them

If you run Fendlemark with the [`kubeflow`](https://fendlemark.app/docs/adapters/#kubeflow), [`argo`](https://fendlemark.app/docs/adapters/#argo), and [`mlflow`](https://fendlemark.app/docs/adapters/#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](https://fendlemark.app/blog/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](https://fendlemark.app/docs/adapters/#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`:

```shell
$ 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:

```yaml
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

- [Adapters: kubeflow](https://fendlemark.app/docs/adapters/#kubeflow), and the [full kubeflow adapter reference](https://fendlemark.app/docs/adapters/kubeflow.md), for every key the adapter reads
- [Running MLflow and Argo Workflows together](https://fendlemark.app/blog/running-mlflow-and-argo-workflows-together/), the same idea for Workflows you write yourself
- [What the words mean in MLflow, Metaflow, Argo Workflows, and Kubeflow](https://fendlemark.app/blog/what-the-words-mean-in-mlflow-metaflow-argo-and-kubeflow/), for why a Kubeflow Pipelines run is also an Argo Workflow
- [Running the Fendlemark app on Kubernetes](https://fendlemark.app/blog/running-the-fendlemark-app-on-kubernetes/), for deploying the console next to Kubeflow
- [Kubeflow Pipelines SDK: `kfp.dsl`](https://kubeflow-pipelines.readthedocs.io/en/stable/source/dsl.html), for the placeholders, and [Kubeflow Pipelines: MLflow integration](https://www.kubeflow.org/docs/components/pipelines/user-guides/integrations/mlflow/overview/)
- [MLflow: Tracking](https://mlflow.org/docs/latest/ml/tracking/), for runs, tags, and `search_runs`