Kubeflow Pipelines: An Introduction for ML Engineers

Kubeflow Pipelines (KFP) is the part of Kubeflow that runs machine learning workflows on Kubernetes. You write each step as a Python function, wire the steps together in another function, and KFP runs every step in its own container, passing files and values between them and recording what each step produced. This introduction builds one small training pipeline, compiles it, and runs it three times on a laptop, including a run where a data check stops the pipeline before a model is trained.

The pieces

• Component: one step, written as a Python function and decorated with @dsl.component, which turns the function into something KFP can run remotely in a container (Kubeflow: lightweight Python components). Imports go inside the function, because only the function's code is shipped to the container.
• Parameters and artifacts: small values such as a date or a threshold are passed as typed parameters. Files such as a dataset or a model are artifacts: a step declares Output[Dataset] and writes to its .path, and the next step declares Input[Dataset] and reads from its .path. KFP decides where the file actually lives.
• Pipeline: a function decorated with @dsl.pipeline that calls components. The order of steps is not written down anywhere; KFP works it out from which outputs feed which inputs.
• Compiled pipeline: the compiler turns the pipeline into Intermediate Representation (IR) YAML, which is what a Kubeflow cluster accepts, and type-checks the connections between steps on the way (Kubeflow: compile a pipeline).

The example

The data are the UCI Beijing PM2.5 records: hourly fine-particle readings at the US Embassy in Beijing from 2010 to 2014, with weather, under CC BY 4.0 (UCI: Beijing PM2.5). The task is to estimate the hour's PM2.5 from the weather and time of day, as a city might do when a sensor is down.

The pipeline has four steps. Ingest downloads the data. Validate refuses the data if too many hours have no reading, then cleans it. Train fits a random forest on everything before a cutoff date. Evaluate scores the model on the following year, next to a baseline that always guesses the training median.

Every step runs in the same image, built from this Dockerfile, so each container has exactly the library versions the code was written against:

FROM python:3.11-slim
RUN pip install --no-cache-dir kfp==2.17.0 pandas==3.0.6 scikit-learn==1.9.1

The pipeline, its compilation, and three local runs:

import contextlib
import io
import re

import yaml
from kfp import compiler, dsl, local
from kfp.dsl import Dataset, Input, Metrics, Model, Output

IMAGE = "kfp-sklearn:1.9.1"     # python:3.11-slim plus kfp 2.17.0, pandas 3.0.6 and scikit-learn 1.9.1 (Dockerfile shown above)


@dsl.component(base_image=IMAGE)
def ingest(raw: Output[Dataset]):
    import io, urllib.request, zipfile
    import pandas as pd
    url = "https://archive.ics.uci.edu/static/public/381/beijing+pm2+5+data.zip"   # UCI Beijing PM2.5 (CC BY 4.0)
    archive = zipfile.ZipFile(io.BytesIO(urllib.request.urlopen(url).read()))
    pd.read_csv(archive.open(archive.namelist()[0])).to_csv(raw.path, index=False)


@dsl.component(base_image=IMAGE)
def validate(raw: Input[Dataset], clean: Output[Dataset], max_missing: float) -> int:
    import pandas as pd
    air = pd.read_csv(raw.path)
    missing = air["pm2.5"].isna().mean()
    if missing > max_missing:
        raise ValueError(f"{missing:.1%} of hours have no PM2.5 reading, above {max_missing:.0%}")
    air = air.dropna(subset=["pm2.5"])
    air["ts"] = pd.to_datetime(air[["year", "month", "day", "hour"]])
    air = pd.get_dummies(air, columns=["cbwd"], dtype=int)            # wind direction as 0/1 columns
    air.drop(columns=["No", "year", "day"]).to_csv(clean.path, index=False)
    return len(air)


@dsl.component(base_image=IMAGE)
def train(clean: Input[Dataset], model: Output[Model], cutoff: str):
    import joblib, pandas as pd
    from sklearn.ensemble import RandomForestRegressor
    air = pd.read_csv(clean.path, parse_dates=["ts"])
    past = air[air.ts < cutoff]
    joblib.dump(RandomForestRegressor(n_estimators=200, min_samples_leaf=5, random_state=0, n_jobs=-1)
                .fit(past.drop(columns=["ts", "pm2.5"]), past["pm2.5"]), model.path)


@dsl.component(base_image=IMAGE)
def evaluate(clean: Input[Dataset], model: Input[Model], start: str, end: str, metrics: Output[Metrics]) -> str:
    import joblib, pandas as pd
    air = pd.read_csv(clean.path, parse_dates=["ts"])
    window, past = air[(air.ts >= start) & (air.ts < end)], air[air.ts < start]
    error = (joblib.load(model.path).predict(window.drop(columns=["ts", "pm2.5"])) - window["pm2.5"]).abs().mean()
    baseline = (past["pm2.5"].median() - window["pm2.5"]).abs().mean()   # always guess the training median
    metrics.log_metric("mae", round(float(error), 1))
    metrics.log_metric("baseline_mae", round(float(baseline), 1))
    return f"MAE {error:.1f} ug/m3 (baseline {baseline:.1f}) on {len(window)} hours"


@dsl.pipeline(name="pm25-training")
def training_pipeline(cutoff: str = "2013-01-01", test_end: str = "2014-01-01", max_missing: float = 0.1) -> str:
    raw = ingest()
    clean = validate(raw=raw.outputs["raw"], max_missing=max_missing)
    model = train(clean=clean.outputs["clean"], cutoff=cutoff)
    return evaluate(clean=clean.outputs["clean"], model=model.outputs["model"], start=cutoff, end=test_end).outputs["Output"]


compiler.Compiler().compile(training_pipeline, "pipeline.yaml")   # this file is what a Kubeflow cluster runs
spec = yaml.safe_load(open("pipeline.yaml"))
print("compiled pipeline.yaml; tasks and what they wait for:")
for name, task in spec["root"]["dag"]["tasks"].items():
    print(f"  {name:<9} after {task.get('dependentTasks', [])}")

local.init(runner=local.DockerRunner(), pipeline_root="/tmp/kfp-outputs")   # each step runs in its own container


def run(label, **params):                                       # run locally, keep KFP's detailed logs out of the way
    log = io.StringIO()
    try:
        with contextlib.redirect_stdout(log):
            result = training_pipeline(**params).output
    except RuntimeError:
        result = None
    plain = re.sub(r"\x1b\[[0-9;]*m", "", log.getvalue())
    result = result or "failed: " + re.findall(r"ValueError: (.*)", plain)[-1]
    steps = ", ".join(f"{task} {status.lower()}" for task, status in re.findall(r"Task '(\w+)' finished with status (\w+)", plain))
    print(f"{label}: {result} | {steps}")


print()
run("run 1, defaults (train 2010-2012, test 2013)")
run("run 2, train 2010-2013, test 2014", cutoff="2014-01-01", test_end="2015-01-01")
run("run 3, stricter data rule (at most 2% missing)", max_missing=0.02)

compiled pipeline.yaml; tasks and what they wait for:
  evaluate  after ['train', 'validate']
  ingest    after []
  train     after ['validate']
  validate  after ['ingest']

run 1, defaults (train 2010-2012, test 2013): MAE 48.5 ug/m3 (baseline 67.6) on 8678 hours | ingest success, validate success, train success, evaluate success
run 2, train 2010-2013, test 2014: MAE 48.0 ug/m3 (baseline 65.4) on 8661 hours | ingest success, validate success, train success, evaluate success
run 3, stricter data rule (at most 2% missing): failed: 4.7% of hours have no PM2.5 reading, above 2% | ingest success, validate failure

What the output shows

• The order came from the data flow. Nothing in the code says "run validate after ingest". The compiled YAML records that validate waits for ingest because it reads ingest's output, and that evaluate waits for both validate and train because it reads the cleaned data and the model. A step with no dependency between them could run at the same time on a cluster.
• Each run used fresh containers. With the local Docker runner, every step started its own container from the base image, read its inputs from disk and wrote its outputs back. That is the same isolation a cluster gives you, which is why a pipeline that works locally is a good sign it will work remotely.
• Parameters made new runs cheap. Run 2 changed two dates and nothing else: the model was trained on 2010 to 2013 and scored on 2014. Its error of 48.0 µg/m³ against a baseline of 65.4 was close to run 1's 48.5 against 67.6, so a year more training data barely helped this weather-only model.
• The data check stopped a bad run early. Run 3 tightened the rule to at most 2% missing readings. The data has 4.7% missing, so validate raised an error, the pipeline failed, and train and evaluate never started. Failing at the cheap step, with a message that says why, is the point of putting validation in its own component. The thinking behind such checks is covered in data quality testing and validation in pipelines.

Running it before you have a cluster

The KFP SDK can run components and pipelines locally: after local.init(), calling the pipeline runs it and returns real values instead of references (Kubeflow: execute pipelines locally). The DockerRunner used above starts a container per step; a SubprocessRunner runs steps in local Python processes instead, which is faster but does not test your images.

The local runner is for development and has limits worth knowing before you rely on it. It does not cache or retry tasks, it ignores resource requests such as memory and accelerators, it has no authentication for cloud resources, and it does not yet support conditions, parallel loops or exit handlers. A pipeline that uses those features has to be tested on a real cluster.

One practical note for Windows: the local Docker runner mounts host folders into each step's container, and Windows paths broke that mount here. Running the driver script itself inside a small Linux container, with the Docker socket mounted and the pipeline root on a Linux path, avoided the problem. On Linux and macOS the script runs directly.

From laptop to cluster

On a cluster, the compiled pipeline.yaml is uploaded through the Kubeflow Pipelines UI or the SDK client, and each run gets a page showing the graph, logs, artifacts and the metrics the evaluate step logged. Two behaviours change. First, caching is on by default: a step whose inputs and parameters match an earlier run reuses that run's outputs instead of executing again (Kubeflow: caching). That is helpful for an expensive training step, and wrong for the ingest step here, which downloads data that could change, so you would turn caching off for it with set_caching_options(False). Second, resource requests start to matter, because each step becomes a pod scheduled onto the cluster's nodes.

The same IR YAML also runs on Google Cloud's managed pipelines service, covered in Vertex AI Pipelines for MLOps on GCP.

Kubeflow Pipelines or something else?

KFP fits when your team already runs Kubernetes and wants each ML step isolated in its own container, with artifacts and metrics tracked per run. If your work is mostly scheduled data jobs, a general orchestrator may fit better; Airflow for ML pipelines shows the same kind of job as a scheduled DAG, and Airflow vs Dagster vs Prefect compares the general-purpose options. If you have not used containers much yet, start with Docker for machine learning.

Common mistakes

• Importing at the top of the file only. A lightweight component runs in its own container, so the imports it needs belong inside the function.
• Passing big data as parameters. Parameters are for small values; datasets and models belong in artifacts.
• Leaving caching on for steps that read changing data, such as a download or a database query.
• Testing only with the subprocess runner, which never builds or uses your container image.
• Putting validation inside the training step, so a bad dataset fails late and with a confusing error.

Building and running pipelines like this one is part of the MLOps course.

Sources

• Lightweight Python Components (Kubeflow documentation) — Kubeflow
• Compile a Pipeline (Kubeflow documentation) — Kubeflow
• Execute KFP pipelines locally (Kubeflow documentation) — Kubeflow
• Use Caching (Kubeflow documentation) — Kubeflow
• Beijing PM2.5 dataset (UCI Machine Learning Repository) — UCI Machine Learning Repository