Deploying an End-to-End Machine Learning Solution on Kubeflow Pipelines

A Kubeflow Pipelines component is the implementation of one task in a pipeline. Each component is a step in the workflow. A component takes inputs, runs its code in a container and produces outputs. Outputs are either small values, called parameters, or files such as datasets and models, called artifacts. A component has an interface (its inputs and outputs), an implementation (a container image and the command it runs) and metadata such as its name.

The Kubeflow Pipelines SDK v2 (the kfp Python package, version 2.x) offers three main ways to write a component:

• Lightweight Python components: a Python function with the @dsl.component decorator. Kubeflow Pipelines runs the function in a base image and installs the packages it lists when the step starts.
• Containerized Python components: Python functions that the kfp component build command packages into a container image with their dependencies.
• Container components: any container image and command, declared with the @dsl.container_component decorator.

See the Kubeflow Pipelines component guides for details on each type.

Overview of a Simple End-to-End Pipeline

In this example, we train a neural network regressor to predict the daily closing price of bitcoin from its opening, high and low prices. The machine learning code is basic because it is not the focus of this part. The goal is to orchestrate a machine learning solution as a set of containerized steps with Kubeflow Pipelines.

The pipeline has four steps:

1. Load the raw dataset into the pipeline.
2. Transform the dataset. Keep the bitcoin rows and the price columns. Split the rows by date: the most recent 20% of days form the evaluation set.
3. Tune the hyperparameters of the model with time-series cross-validation.
4. Train the model with the best hyperparameters. Report its error on the evaluation set beside a baseline that predicts each day closes at its opening price.

Splitting by date matters for a price series. A random split lets the model train on days that come after the days it is tested on, which makes its error look lower than it will be in use.

The 2019 version of this pipeline used Cloud Dataflow for the transform step and Cloud Machine Learning Engine for tuning, training and serving. Its components called the gcloud ml-engine commands and trained a TensorFlow 1.8 model. Cloud Machine Learning Engine later became AI Platform and then Vertex AI. In April 2026 Google introduced Gemini Enterprise Agent Platform as the successor to Vertex AI. The pipeline below does its work inside lightweight Python components with pandas and scikit-learn. It runs the same way on any backend that supports the KFP SDK.

The Dataset

The pipeline reads a CSV file of daily cryptocurrency prices from 28 April 2013 to 10 January 2018. It covers 971 coins in 498,380 rows. The file is in the source repository at kubeflow-pipelines/data/crypto-markets.csv.gz, and the pipeline reads it from there by default. To use another file, pass its location as the data_url parameter. The location can be an HTTPS URL, a local path for local tests or a Cloud Storage path such as gs://[BUCKET_NAME]/crypto-markets.csv. The file needs the columns slug, date, open, high, low and close. The slug column holds the name of the coin, such as bitcoin.

Container Images and Credentials

In the 2019 version, each component was a separate Docker image. Each image held its code and a service account key file for authenticating to Google Cloud. Do not put key files in container images. Anyone who can pull the image can read the key. Give the workload an identity instead:

• On GKE, use Workload Identity Federation for GKE. It lets Kubernetes service accounts act as IAM principals without key files.
• On Agent Platform Pipelines, each run uses a Google Cloud service account. If you do not choose one, the run uses the Compute Engine default service account.

Lightweight Python components need no image builds. If you write container components with your own images, push the images to Artifact Registry. The 2019 version pushed its images to Container Registry. Google shut down Container Registry on March 18, 2025.

Install the Kubeflow Pipelines SDK

The SDK needs Python 3.9 or later. Install it with pip:

bashCopy
# install the kubeflow pipelines sdk
pip install kfp

# check the installed version
python -c "import kfp; print(kfp.__version__)"

At the time of writing, the latest version of the SDK is 2.17.0. The code below was tested with that version.

Write the Pipeline

The code below is kubeflow-pipelines/crypto_pipeline.py in the source repository. Each function with the @dsl.component decorator is one step. The @dsl.pipeline function connects the steps. The SDK works out the order of the steps from the way outputs feed into inputs.

pythonCopy
"""Kubeflow for Poets: an end-to-end pipeline that predicts bitcoin's daily close.

The pipeline has four steps:

1. Load the raw CSV of daily cryptocurrency prices.
2. Keep the bitcoin rows and split them by date into training and evaluation sets.
3. Tune a neural network regressor with time-series cross-validation.
4. Train the model with the best settings and report its error on the evaluation set.

Prices in later years are far above those in earlier years, so the model works with
ratios to the day's opening price. It learns close/open from high/open and low/open,
then multiplies by the open to get a price. This lets it handle price levels it never
saw in training.

Compile it with `python crypto_pipeline.py`. The result, crypto_pipeline.yaml, runs on
Kubeflow Pipelines and on Agent Platform Pipelines.
"""

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

# The dataset this series uses, kept in this repository.
DATA_URL = (
    "https://raw.githubusercontent.com/dvdbisong/kubeflow-for-poets/master/"
    "kubeflow-pipelines/data/crypto-markets.csv.gz"
)

BASE_IMAGE = "python:3.12-slim"
PANDAS = "pandas==3.0.6"
SKLEARN = "scikit-learn==1.9.1"


@dsl.component(base_image=BASE_IMAGE, packages_to_install=[PANDAS, "gcsfs"])
def load_raw_data(data_url: str, raw_data: Output[Dataset]):
    """Copy the raw CSV into the pipeline's artifact store.

    data_url can be an HTTPS URL, a local path or a Cloud Storage path (gs://...).
    Compressed files such as .csv.gz are read as they are.
    """
    import pandas as pd

    df = pd.read_csv(data_url)
    required = {"slug", "date", "open", "high", "low", "close"}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"the dataset is missing columns: {sorted(missing)}")
    df.to_csv(raw_data.path, index=False)


@dsl.component(base_image=BASE_IMAGE, packages_to_install=[PANDAS])
def transform_data(
    raw_data: Input[Dataset],
    train_data: Output[Dataset],
    eval_data: Output[Dataset],
    coin: str = "bitcoin",
    eval_fraction: float = 0.2,
):
    """Keep one coin's price columns and split the rows by date.

    The most recent days form the evaluation set, so the model never trains on
    days that come after the ones it is tested on.
    """
    import pandas as pd

    df = pd.read_csv(raw_data.path, parse_dates=["date"])
    df = df[df["slug"] == coin][["date", "open", "high", "low", "close"]]
    df = df.dropna().sort_values("date").reset_index(drop=True)
    if len(df) < 50:
        raise ValueError(f"only {len(df)} rows for {coin}. The pipeline needs at least 50.")

    cut = int(len(df) * (1 - eval_fraction))
    df.iloc[:cut].to_csv(train_data.path, index=False)
    df.iloc[cut:].to_csv(eval_data.path, index=False)


@dsl.component(base_image=BASE_IMAGE, packages_to_install=[PANDAS, SKLEARN])
def tune_hyperparameters(train_data: Input[Dataset], metrics: Output[Metrics]) -> dict:
    """Search learning rates and layer sizes with time-series cross-validation."""
    import pandas as pd
    from sklearn.model_selection import GridSearchCV, TimeSeriesSplit
    from sklearn.neural_network import MLPRegressor
    from sklearn.pipeline import make_pipeline
    from sklearn.preprocessing import StandardScaler

    df = pd.read_csv(train_data.path)
    # Ratios to the opening price keep the inputs on the same scale in every year.
    features = pd.DataFrame({"high": df["high"] / df["open"], "low": df["low"] / df["open"]})
    target = df["close"] / df["open"]
    search = GridSearchCV(
        make_pipeline(StandardScaler(), MLPRegressor(max_iter=1000, random_state=42)),
        param_grid={
            "mlpregressor__learning_rate_init": [0.001, 0.01],
            "mlpregressor__hidden_layer_sizes": [(64,), (128, 64), (256, 128, 64)],
        },
        scoring="neg_root_mean_squared_error",
        cv=TimeSeriesSplit(n_splits=3),
    )
    search.fit(features, target)
    metrics.log_metric("cv_rmse_ratio", float(-search.best_score_))

    best = search.best_params_
    return {
        "learning_rate_init": best["mlpregressor__learning_rate_init"],
        "hidden_layer_sizes": list(best["mlpregressor__hidden_layer_sizes"]),
    }


@dsl.component(base_image=BASE_IMAGE, packages_to_install=[PANDAS, SKLEARN, "joblib"])
def train_model(
    train_data: Input[Dataset],
    eval_data: Input[Dataset],
    best_params: dict,
    model: Output[Model],
    metrics: Output[Metrics],
):
    """Train with the best settings and report RMSE and MAE in dollars on the evaluation set.

    The baseline metrics predict that each day closes at its opening price.
    """
    import json
    import os

    import joblib
    import pandas as pd
    from sklearn.metrics import mean_absolute_error, root_mean_squared_error
    from sklearn.neural_network import MLPRegressor
    from sklearn.pipeline import make_pipeline
    from sklearn.preprocessing import StandardScaler

    train = pd.read_csv(train_data.path)
    evaluation = pd.read_csv(eval_data.path)

    def ratios(df):
        return pd.DataFrame({"high": df["high"] / df["open"], "low": df["low"] / df["open"]})

    regressor = make_pipeline(
        StandardScaler(),
        MLPRegressor(
            learning_rate_init=best_params["learning_rate_init"],
            # A dict output stores numbers as floats, so convert the sizes back to int.
            hidden_layer_sizes=tuple(int(n) for n in best_params["hidden_layer_sizes"]),
            max_iter=1000,
            random_state=42,
        ),
    )
    regressor.fit(ratios(train), train["close"] / train["open"])
    predictions = regressor.predict(ratios(evaluation)) * evaluation["open"]

    actual = evaluation["close"]
    metrics.log_metric("rmse", float(root_mean_squared_error(actual, predictions)))
    metrics.log_metric("mae", float(mean_absolute_error(actual, predictions)))
    metrics.log_metric("baseline_rmse", float(root_mean_squared_error(actual, evaluation["open"])))
    metrics.log_metric("baseline_mae", float(mean_absolute_error(actual, evaluation["open"])))

    os.makedirs(model.path, exist_ok=True)
    joblib.dump(regressor, os.path.join(model.path, "model.joblib"))
    with open(os.path.join(model.path, "features.json"), "w") as f:
        json.dump(
            {
                "inputs": ["open", "high", "low"],
                "features": ["high / open", "low / open"],
                "target": "close / open",
                "prediction": "open * model output",
            },
            f,
        )
    model.metadata["framework"] = "scikit-learn"


@dsl.pipeline(name="crypto", description="Train a model that predicts bitcoin's daily closing price")
def crypto_pipeline(data_url: str = DATA_URL, coin: str = "bitcoin"):
    raw = load_raw_data(data_url=data_url)
    transformed = transform_data(raw_data=raw.outputs["raw_data"], coin=coin)
    tuned = tune_hyperparameters(train_data=transformed.outputs["train_data"])
    train_model(
        train_data=transformed.outputs["train_data"],
        eval_data=transformed.outputs["eval_data"],
        best_params=tuned.outputs["Output"],
    )


if __name__ == "__main__":
    compiler.Compiler().compile(pipeline_func=crypto_pipeline, package_path="crypto_pipeline.yaml")
    print("Wrote crypto_pipeline.yaml")

A few points about the code:

• Each component imports its own libraries inside the function body. The function runs in its own container and cannot see anything defined outside it.
• Output[Dataset] and Output[Model] give a component a local path to write an artifact to. Kubeflow Pipelines then copies the file to the artifact store. Input[Dataset] gives the next component a local path to read it from.
• Bitcoin traded at far higher prices in late 2017 than in any earlier year. A model trained on raw prices cannot predict prices it has never seen. The model works with ratios to the opening price instead. It learns close/open from high/open and low/open, then multiplies by the open to get a price.
• tune_hyperparameters returns a dict and also logs a metric. A component with more than one output names its return value Output, so the pipeline passes it to train_model as tuned.outputs["Output"].
• metrics.log_metric records the root mean squared error (RMSE) and the mean absolute error (MAE) in dollars, with the same two numbers for the baseline.

Test the Pipeline Locally

The SDK can run a pipeline on your machine without a cluster. The subprocess runner creates a virtual environment for each step and installs the packages that the step lists. The source repository has a script that does this:

bashCopy
# get the code and install the sdk
git clone https://github.com/dvdbisong/kubeflow-for-poets.git
cd kubeflow-for-poets
python -m venv .venv
source .venv/bin/activate
pip install -r requirements-dev.txt

# run the whole pipeline locally
cd kubeflow-pipelines
python run_local.py

The script calls the pipeline like this:

pythonCopy
from kfp import local

from crypto_pipeline import crypto_pipeline

# run each step in a local virtual environment
local.init(runner=local.SubprocessRunner(use_venv=True), pipeline_root="local_outputs")

# run the pipeline on the copy of the dataset in the repository
crypto_pipeline(data_url="data/crypto-markets.csv.gz")

The log shows each task as it starts and finishes. On the evaluation set, the most recent 344 days in the data, the model's RMSE is about 183 dollars. The baseline that predicts each day closes at its open has an RMSE of about 447 dollars. The repository also has tests that run every step on a small synthetic dataset. Run them with pytest. See Execute KFP pipelines locally for the other runner options.

Compile the Pipeline

The compiler turns the pipeline into an IR YAML file that a Kubeflow Pipelines backend can run. The 2019 version of the SDK compiled pipelines to a .tar.gz archive. The SDK v2 compiles them to YAML.

bashCopy
# compile the pipeline to crypto_pipeline.yaml
python crypto_pipeline.py

The kfp command-line tool does the same thing:

bashCopy
# compile the pipeline with the kfp cli
kfp dsl compile --py crypto_pipeline.py --output crypto_pipeline.yaml

Run the Pipeline on Kubeflow Pipelines

You can run the compiled pipeline from the UI or from Python. By default the pipeline pods read the dataset from the source repository on GitHub. If you point data_url at a Cloud Storage bucket, grant the pods access with Workload Identity Federation for GKE.

From the UI

1. Forward a local port to the UI as shown in the previous part. Open the UI in a browser.
2. Click Pipelines. Then click Upload pipeline. Choose crypto_pipeline.yaml, give the pipeline a name and click Create.
3. Click the pipeline to see the graph of its four steps.
4. Click Runs. Then click Create run. Select the pipeline, choose or create an experiment and click Start. Change data_url only if you want to use another copy of the dataset.
5. Click the run to follow its progress. Click a step to see its logs and artifacts. The metrics artifact of the train-model step holds the RMSE and the MAE.

From Python

For a standalone installation of Kubeflow Pipelines, forward a local port to the ml-pipeline-ui service:

bashCopy
# forward local port 3000 to the kubeflow pipelines ui
kubectl port-forward -n kubeflow svc/ml-pipeline-ui 3000:80

Then submit a run with the kfp.Client class:

pythonCopy
import kfp

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

run = client.create_run_from_pipeline_package(
    "crypto_pipeline.yaml",
    arguments={"data_url": "gs://[BUCKET_NAME]/crypto-markets.csv"},
    experiment_name="crypto",
)
print(run.run_id)

A full Kubeflow installation requires a login before the client can submit runs. See Connect the SDK to the API for the steps for each type of installation.

Run the Pipeline on Agent Platform Pipelines

The same crypto_pipeline.yaml file runs on Agent Platform Pipelines without a Kubernetes cluster. The service stores the artifacts of each run in a Cloud Storage bucket. See Configure your project for the full setup and the roles that the service account needs.

bashCopy
# enable the required apis
gcloud services enable aiplatform.googleapis.com compute.googleapis.com storage-component.googleapis.com

# create a bucket for the pipeline artifacts
gcloud storage buckets create gs://[BUCKET_NAME] --location=us-central1

# install the google cloud sdk for python
pip install google-cloud-aiplatform

Submit the run with the PipelineJob class:

pythonCopy
from google.cloud import aiplatform

job = aiplatform.PipelineJob(
    display_name="crypto",
    template_path="crypto_pipeline.yaml",
    pipeline_root="gs://[BUCKET_NAME]/pipeline-root",
    parameter_values={"data_url": "gs://[BUCKET_NAME]/crypto-markets.csv"},
    project="[PROJECT_ID]",
    location="us-central1",
)
job.submit()

Follow the run on the Pipelines page of the Agent Platform section in the Google Cloud console. The page shows the same graph of four steps with their logs, artifacts and metrics.

Serving the Model

The train_model step saves the trained model as model.joblib in the artifact store. To serve predictions from it:

• On Kubeflow, deploy the model with KServe. The Kubeflow Community Distribution installs it.
• On Google Cloud, import the model into the Agent Platform Model Registry and deploy it to an endpoint for online predictions.

Delete Resources

Delete the GKE Cluster

Deleting the cluster also deletes Kubeflow or Kubeflow Pipelines and everything else that runs on it. Use the name and location of the cluster you created in an earlier part.

bashCopy
# delete the cluster from the kubeflow part
gcloud container clusters delete kubeflow-cluster --location=us-central1-a

# delete the cluster from the kubeflow pipelines part
gcloud container clusters delete kfp-cluster --location=us-central1-a

Delete the Cloud Storage Bucket

Agent Platform Pipelines has no cluster to delete. Its runs leave artifacts in the bucket. This command deletes the bucket and every object in it:

bashCopy
# delete the bucket and its contents
gcloud storage rm --recursive gs://[BUCKET_NAME]

References

The following resources help with creating and running Kubeflow pipelines:

• Kubeflow Pipelines overview
• Compile a pipeline
• Run a pipeline
• Kubeflow Pipelines SDK reference
• Introduction to Agent Platform Pipelines
• Source code for the Kubeflow for Poets series