Kubeflow Pipelines on GKE

Deploy Kubeflow Pipelines on GKE and run a multi-step ML pipeline.

Introduction

30 Second Summary

Machine learning models don't just need good code. They need a reliable system to move data through preparation, training, and evaluation steps without manual handholding.

In this project, you will deploy Kubeflow Pipelines on a Google Kubernetes Engine cluster and build a multi-step ML pipeline that processes data, trains a model, and evaluates results. Every step runs as its own Kubernetes pod.

What You'll Build

You'll watch your Python pipeline definition transform into a live DAG of Kubernetes pods, with each step's logs, inputs, and outputs visible in the Kubeflow Pipelines dashboard as they execute in real time.

By the end of this project, you'll have:

  • A GKE cluster running Kubeflow Pipelines with the full web dashboard accessible via port-forward, ready to accept and run your pipelines.
  • A Python pipeline script that defines four ML components chained together as a DAG, compiled to YAML and executed through the KFP UI.
  • A clear mental model of how KFP maps your Python decorators to Kubernetes pods, Argo Workflows specs, and pipeline artifacts.
  • Secret Mission: Add conditional branching that only evaluates your model when its training accuracy exceeds a threshold.

Are there any prerequisites?

This project assumes you have Kubernetes fundamentals (CKA/CKAD level) and basic Python experience. You'll also need a free KodeKloud account to access their GCP Playground, which provides a time-limited Google Cloud Shell environment with all the tools pre-installed.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters for your ML engineering toolkit.

Create a GKE Cluster

Every ML pipeline platform needs compute underneath it. Kubeflow Pipelines runs on Kubernetes, so the first thing you need is a cluster.

In this step, you will create a GKE Standard cluster inside the KodeKloud GCP Playground and configure kubectl to talk to it.

In this step, get ready to:
  • Create a 3-node GKE Standard cluster in the us-west1 region.
  • Configure kubectl credentials to connect to the cluster.
  • Verify that all three nodes are in Ready status.
Create the GKE cluster

The KodeKloud GCP Playground supports GKE Standard clusters with e2-medium nodes, up to 3 nodes, in the us-west1 region.

  • Create a GKE cluster named kfp-cluster by running this command in Cloud Shell:
gcloud container clusters create kfp-cluster \
  --location us-west1-b \
  --num-nodes 3 \
  --machine-type e2-medium

What does this command do?

This single command provisions a full Kubernetes cluster on Google Cloud.

  • kfp-cluster is the name you are giving the cluster.
  • --location us-west1-b places the cluster's control plane and nodes in the us-west1-b zone (inside the us-west1 region that KodeKloud allows).
  • --num-nodes 3 creates three worker nodes for the cluster.
  • --machine-type e2-medium selects a cost-effective VM type with 2 vCPUs and 4 GB RAM per node.

This takes a few minutes

Cluster creation typically takes 5 to 8 minutes. You will see progress messages in Cloud Shell while GKE provisions your control plane and nodes. Wait for the command to finish before continuing.

When the command completes, you will see a table in the terminal showing your cluster's name, location, version, and node count.

Configure kubectl and verify the cluster

The cluster exists, but kubectl does not know about it yet. You need to fetch credentials so kubectl points at your new cluster.

  • Configure kubectl to connect to your cluster by running:
gcloud container clusters get-credentials kfp-cluster --location us-west1-b

What does this command do?

This command updates your kubeconfig file with the endpoint and authentication details for kfp-cluster. After running it, every kubectl command you type targets this cluster.

Before you run the next command, take a moment to predict: how many nodes do you expect to see, and what status should they be in?

  • Verify your cluster is healthy by running:
kubectl get nodes

You should see three nodes listed, each with a STATUS of Ready. That confirms your cluster is up and accepting workloads.

Not seeing three Ready nodes?

  • If you see fewer than three nodes, the cluster may still be provisioning. Wait a minute and run kubectl get nodes again.
  • If a node shows NotReady, give it another minute. Nodes can take a short while to finish registering after the create command returns.
  • If the command says the connection was refused, re-run gcloud container clusters get-credentials kfp-cluster --location us-west1-b and try again.

Still stuck? Ask for help.

Your cluster is live and ready to host workloads. Next up, you will deploy Kubeflow Pipelines onto this cluster and open the web UI.

Deploy Kubeflow Pipelines and Access the UI

Your GKE cluster is running with three healthy nodes. A bare Kubernetes cluster has no concept of ML pipeline stages, experiment tracking, or artifact management.

Kubeflow Pipelines adds the missing orchestration layer. In this step, you will deploy it onto your cluster using official kustomize manifests and access its web dashboard.

In this step, get ready to:
  • Deploy Kubeflow Pipelines 2.17.0 using official kustomize manifests.
  • Verify all KFP pods are running in the kubeflow namespace.
  • Access the KFP web UI through Cloud Shell Web Preview.
Deploy the Kubeflow Pipelines manifests

The deployment happens in two phases. First you apply cluster-scoped resources like CRDs and permissions. Then you apply the namespaced components that make up the platform.

  • Set the pipeline version and apply the cluster-scoped resources by running these commands in your Cloud Shell:
export PIPELINE_VERSION=2.17.0
kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/cluster-scoped-resources?ref=$PIPELINE_VERSION"

You will see a list of created resources in the output, including CRDs and cluster roles.

  • Wait for the CRDs to be fully registered before continuing by running:
kubectl wait --for condition=established --timeout=60s crd/applications.app.k8s.io

Once you see a confirmation that the condition is met, the cluster is ready for the next phase.

  • Deploy the Kubeflow Pipelines dev environment by running:
kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/env/dev?ref=$PIPELINE_VERSION"

What do these commands do?

The first command sets a shell variable so every manifest URL points to version 2.17.0. The kubectl apply -k commands pull kustomize manifests directly from the Kubeflow GitHub repository and apply them to your cluster.

The env/dev flavor is labeled "non-production for first experiments" in the official docs. It is a lightweight deployment perfect for learning, without the overhead of a production setup.

Seeing a timeout or connection error?

The kubectl apply -k command downloads manifests from GitHub. If your Cloud Shell session has intermittent connectivity, wait a few seconds and re-run the same command. It is safe to re-apply.

Help me troubleshoot this connection issue

Verify all KFP pods are running

The deployment takes approximately 3 minutes. The manifests create a kubeflow namespace and launch several pods inside it.

  • Check the status of all pods in the kubeflow namespace by running:
kubectl get pods -n kubeflow
  • If some pods show ContainerCreating or PodInitializing, wait about a minute and re-run the command.
  • All pods should eventually show Running in the STATUS column.

What are these pods?

Each pod is a component of the Kubeflow Pipelines platform. Here are the key ones you will see:

  • ml-pipeline is the API server. It accepts pipeline definitions, creates runs, and manages experiment metadata.
  • ml-pipeline-ui is the web dashboard you will access shortly to visualize pipeline DAGs, logs, and artifacts.
  • mysql is the metadata store. It persists pipeline definitions, run history, and experiment records.
  • minio is the artifact storage layer. Pipeline steps write outputs here so downstream steps can read them.
  • workflow-controller is the Argo Workflows engine. It translates your pipeline DAG into actual Kubernetes pods that execute each step.

Pods stuck in Pending or ContainerCreating?

If pods remain stuck after 5 minutes, check the events for a specific pod by running the describe command below and looking at the Events section at the bottom. Common causes on GKE include image pull delays and resource scheduling.

kubectl describe pod <pod-name> -n kubeflow

Replace <pod-name> with the name of the stuck pod from the kubectl get pods output.

Help me diagnose stuck KFP pods

Access the KFP web UI through port-forwarding

The KFP dashboard runs inside the cluster with no public endpoint. Port-forwarding maps the in-cluster service to a local port so you can reach it from your browser.

  • Start port-forwarding the KFP UI service by running:
kubectl port-forward -n kubeflow svc/ml-pipeline-ui 8080:80

What is happening in the terminal?

You will see a forwarding message and the terminal will stay occupied. This is expected. The port-forward must keep running for the UI to be accessible.

Do not close this terminal or press Ctrl+C. You will open the dashboard from the Cloud Shell toolbar instead.

Before you open the preview, what do you expect to see in the KFP dashboard?

  • Click the Web preview button in the top-right area of the Cloud Shell toolbar.
  • Select Preview on port 8080.

A new browser tab opens with the Kubeflow Pipelines dashboard. You are now looking at the platform that will orchestrate your ML workflows as Kubernetes-native DAGs.

Open a new Cloud Shell tab for future commands

The port-forward is still running in your current terminal. To continue working, click the + icon on the Cloud Shell menu bar to open a new terminal tab. Use that new tab for all commands in the next steps.

Seeing a blank page or connection refused?

Make sure the port-forward terminal is still running. If it stopped, re-run the kubectl port-forward command above.

If the page loads but shows an error, confirm all pods in the kubeflow namespace are in Running status. The ml-pipeline-ui pod must be fully ready for the dashboard to respond.

Help me fix the KFP UI connection issue

Your cluster is now a fully equipped ML platform. Next up, you will write your first KFP pipeline in Python, compile it, and watch Kubeflow turn it into running Kubernetes pods.

Write Your First Pipeline and Watch It Fail

Your Kubeflow Pipelines deployment is running and the dashboard is loaded. The infrastructure is ready, but right now it has no pipelines of your own.

The fastest way to understand KFP components is to write one, compile it, and watch Kubeflow turn your Python function into a Kubernetes pod. In this step, you'll write a two-component pipeline, run it through the KFP UI, and then deliberately break it to see how Kubeflow surfaces errors.

In this step, get ready to:
  • Install the KFP Python SDK in Cloud Shell.
  • Write, compile, and run a two-component hello pipeline.
  • Break the pipeline and observe failure handling in the KFP UI.
Install the KFP SDK

The KFP Python SDK is the library that lets you define pipeline components as Python functions and compile them into Kubernetes-ready YAML files.

  • Your current Cloud Shell tab is running the port-forward command from the previous step. Open a new terminal tab by clicking the + icon next to your current Cloud Shell session tab.
  • Install the KFP SDK by running this command:
pip install kfp==2.17.0

The install takes about a minute. Once it finishes, confirm the version is correct.

  • Verify the installation by running:
pip show kfp | grep Version

You should see Version: 2.17.0 in the output.

Not seeing the right version?

Make sure you typed kfp==2.17.0 with two equals signs. A single equals sign causes a syntax error.

If pip is not found, confirm you are in the new Cloud Shell tab (not the one running port-forward). Cloud Shell has pip pre-installed alongside Python 3.12.

help me troubleshoot the KFP SDK installation in Cloud Shell

Write and compile a hello pipeline

Each KFP component is a Python function decorated with @dsl.component. KFP packages each component into its own container image and runs it as a separate Kubernetes pod.

  • Create a new file called hello_pipeline.py by running:
touch hello_pipeline.py
  • Confirm the file was created by running:
ls hello_pipeline.py

You should see hello_pipeline.py in the output.

  • Open the Cloud Shell Editor by clicking the Open Editor button (pencil icon) at the top of Cloud Shell.
  • In the editor's file explorer, open hello_pipeline.py and add the following code for your two pipeline components:
from kfp import dsl
from kfp import compiler


# Component 1: generate a greeting message
@dsl.component(base_image='python:3.12-slim')
def say_hello(name: str) -> str:
    hello_text = f'Hello, {name}!'
    print(hello_text)
    return hello_text


# Component 2: receive and print a message
@dsl.component(base_image='python:3.12-slim')
def print_message(message: str):
    print(f'Received message: {message}')

What does this code do?

  • from kfp import dsl and from kfp import compiler bring in the decorators and compiler you need to define and compile pipelines.
  • @dsl.component(base_image='python:3.12-slim') tells KFP to run this function inside a python:3.12-slim container. Each decorated function becomes its own Kubernetes pod at runtime.
  • say_hello takes a name string, builds a greeting, and returns it as a typed output that other components can consume.
  • print_message receives a string and prints it. It has no return value.
  • Add the pipeline definition and compilation logic below the components in hello_pipeline.py:
# Wire the components into a pipeline DAG
@dsl.pipeline(name='hello-pipeline')
def hello_pipeline(recipient: str = 'Kubeflow'):
    hello_task = say_hello(name=recipient)
    print_message(message=hello_task.output)


if __name__ == '__main__':
    compiler.Compiler().compile(hello_pipeline, 'hello_pipeline.yaml')
    print('Pipeline compiled to hello_pipeline.yaml')

What does the pipeline definition do?

  • @dsl.pipeline wires the components into a DAG. hello_task.output passes the return value of say_hello as input to print_message. KFP uses this dependency to determine execution order.
  • The if __name__ block compiles the pipeline definition into a YAML file that the KFP backend can execute.
  • Save hello_pipeline.py in the editor.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before continuing.

ⓧ I'd like to double check the full code

Your hello_pipeline.py should look exactly like this:

from kfp import dsl
from kfp import compiler


@dsl.component(base_image='python:3.12-slim')
def say_hello(name: str) -> str:
    hello_text = f'Hello, {name}!'
    print(hello_text)
    return hello_text


@dsl.component(base_image='python:3.12-slim')
def print_message(message: str):
    print(f'Received message: {message}')


@dsl.pipeline(name='hello-pipeline')
def hello_pipeline(recipient: str = 'Kubeflow'):
    hello_task = say_hello(name=recipient)
    print_message(message=hello_task.output)


if __name__ == '__main__':
    compiler.Compiler().compile(hello_pipeline, 'hello_pipeline.yaml')
    print('Pipeline compiled to hello_pipeline.yaml')
  • Switch back to your new Cloud Shell terminal tab (not the port-forward tab). Click Open Terminal if you are still in the editor view.
  • Compile the pipeline by running:
python hello_pipeline.py

You should see Pipeline compiled to hello_pipeline.yaml printed in the terminal.

  • Peek at the compiled YAML to see the structure KFP generated by running:
head -40 hello_pipeline.yaml

What is inside the compiled YAML?

The YAML is a KFP v2 Intermediate Representation (IR) that describes your pipeline as a DAG. You will recognize the structure from your Kubernetes background.

  • deploymentSpec.executors defines the container specs for each component. Each executor maps to a pod with the python:3.12-slim image.
  • root.dag.tasks lists the DAG tasks and their dependencies, wiring say_hello's output into print_message's input.
  • At runtime, the Argo Workflows engine reads this spec and creates one pod per task.

Seeing a compilation error?

Check that your @dsl.component decorators have matching parentheses and that your function signatures include type annotations (e.g. name: str). KFP requires type hints on all component inputs and outputs.

help me fix a KFP pipeline compilation error

Run the pipeline, then break it

The compiled YAML is ready to submit. You will upload it to the KFP UI, watch it succeed, then create a broken version to see how Kubeflow handles failures.

  • Download hello_pipeline.yaml to your local machine by running this command in Cloud Shell:
cloudshell download hello_pipeline.yaml

Why download first?

The KFP UI runs in your browser via Web Preview, but its file upload dialog browses your local machine's filesystem. Since the YAML file lives on the Cloud Shell VM, you need to download it locally before you can upload it through the browser.

  • Switch to the browser tab showing the KFP UI (your Web Preview on port 8080).
  • Click Pipelines in the left sidebar.
  • Click + Upload pipeline.
  • Select hello_pipeline.yaml from your local downloads folder. Fill in a pipeline name (e.g. hello-pipeline) and click Create.
  • Click + Create run on the pipeline detail page.
  • Leave the default parameter values and click Start.

The run starts executing. You will see a DAG graph with two nodes. Within a minute or two, both nodes should turn green with checkmarks.

  • Click the say-hello node in the graph to inspect its logs. You should see Hello, Kubeflow! in the output.
  • Click the print-message node. You should see Received message: Hello, Kubeflow! in the output.

What just happened under the hood?

KFP read your compiled YAML, created an Argo Workflow, and scheduled two pods on your GKE cluster. The first pod ran say_hello inside a python:3.12-slim container. Its return value was passed as an input artifact to the second pod running print_message. Each Python function became a standalone Kubernetes pod with its own lifecycle.

Now you will deliberately break the pipeline to see how KFP handles errors.

  • Switch back to your Cloud Shell terminal tab.
  • Create a new file called broken_pipeline.py by running:
touch broken_pipeline.py
  • Open the Cloud Shell Editor and add the following code to broken_pipeline.py. This is the same pipeline with one key change. The print_message component now raises a ValueError instead of printing the message:
from kfp import dsl
from kfp import compiler


# Same greeting component as before
@dsl.component(base_image='python:3.12-slim')
def say_hello(name: str) -> str:
    hello_text = f'Hello, {name}!'
    print(hello_text)
    return hello_text


# This component deliberately raises an error
@dsl.component(base_image='python:3.12-slim')
def print_message(message: str):
    raise ValueError(f'Intentional error to demonstrate failure handling: {message}')
  • Add the pipeline definition and compilation logic below:
@dsl.pipeline(name='broken-pipeline')
def broken_pipeline(recipient: str = 'Kubeflow'):
    hello_task = say_hello(name=recipient)
    print_message(message=hello_task.output)


if __name__ == '__main__':
    compiler.Compiler().compile(broken_pipeline, 'broken_pipeline.yaml')
    print('Pipeline compiled to broken_pipeline.yaml')
  • Save broken_pipeline.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before continuing.

ⓧ I'd like to double check the full code

Your broken_pipeline.py should look exactly like this:

from kfp import dsl
from kfp import compiler


@dsl.component(base_image='python:3.12-slim')
def say_hello(name: str) -> str:
    hello_text = f'Hello, {name}!'
    print(hello_text)
    return hello_text


@dsl.component(base_image='python:3.12-slim')
def print_message(message: str):
    raise ValueError(f'Intentional error to demonstrate failure handling: {message}')


@dsl.pipeline(name='broken-pipeline')
def broken_pipeline(recipient: str = 'Kubeflow'):
    hello_task = say_hello(name=recipient)
    print_message(message=hello_task.output)


if __name__ == '__main__':
    compiler.Compiler().compile(broken_pipeline, 'broken_pipeline.yaml')
    print('Pipeline compiled to broken_pipeline.yaml')
  • Switch to the Cloud Shell terminal tab and compile the broken pipeline by running:
python broken_pipeline.py

You should see Pipeline compiled to broken_pipeline.yaml in the terminal. The pipeline compiles successfully because the error only happens at runtime, not at compile time.

Before you upload this, what do you think will happen when the pipeline runs? Which step will fail, and which will succeed?

  • Download broken_pipeline.yaml to your local machine by running:
cloudshell download broken_pipeline.yaml
  • Switch to the KFP UI browser tab.
  • Click Pipelines in the left sidebar, then click + Upload pipeline.
  • Upload broken_pipeline.yaml from your local downloads. Name it broken-pipeline and click Create.
  • Click + Create run, leave defaults, and click Start.

Watch the DAG graph. The say-hello node turns green (it succeeded). The print-message node turns red (it failed with your intentional ValueError).

  • Click the red print-message node. Open the Logs tab to see the full error traceback, including your Intentional error to demonstrate failure handling message.

Why break things on purpose?

In production ML pipelines, individual steps fail regularly. Data sources go missing, models fail to converge, containers run out of memory. Knowing where to find error logs, which step failed, and what the upstream steps did is essential for debugging.

KFP shows you all of this in one DAG view. Green nodes succeeded, red nodes failed, and you can click any node to inspect its inputs, outputs, and logs.

Both steps show green instead of a red error?

Make sure you uploaded broken_pipeline.yaml (not hello_pipeline.yaml). Check the pipeline name in the UI. It should say broken-pipeline.

Also confirm that broken_pipeline.py has raise ValueError in the print_message function, not print(). If you fixed the file, recompile with python broken_pipeline.py, download the new YAML, and upload it again.

help me debug why my broken pipeline is not showing an error in the KFP UI

You now have a working pipeline and a broken one in your KFP dashboard. You have seen how Kubeflow turns Python functions into pods, passes data between them, and surfaces failures. Next up, you will build a real multi-step ML pipeline that chains data through four stages.

Build a Multi-Step ML Pipeline

You have proven that KFP works. Your two-step pipelines ran as Kubernetes pods, and you saw firsthand how KFP surfaces errors in the UI.

A real ML workflow needs more than two steps. Data flows through a chain of stages: loading, preprocessing, training, and evaluation. In this step, you will build a four-component pipeline where each component's output feeds into the next, creating a DAG that KFP executes across separate pods.

In this step, get ready to:
  • Write four KFP components that form an ML training pipeline.
  • Wire the components into a DAG using the @dsl.pipeline decorator.
  • Compile, upload, and run the pipeline in the KFP UI.
Write the data loading and preprocessing components

The first two components handle data. One generates a synthetic CSV dataset. The other normalizes the feature values using min-max scaling.

  • Create a new file called ml_pipeline.py in the Cloud Shell editor. Add the imports and the load_data component by pasting this code:
from kfp import dsl
from kfp import compiler


@dsl.component(base_image='python:3.12-slim')
def load_data() -> str:
    import csv
    import io
    import random

    random.seed(42)
    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(['feature1', 'feature2', 'feature3', 'label'])

    for _ in range(100):
        f1 = round(random.uniform(0, 10), 2)
        f2 = round(random.uniform(0, 10), 2)
        f3 = round(random.uniform(0, 10), 2)
        label = 1 if (f1 + f2) > 10 else 0
        writer.writerow([f1, f2, f3, label])

    csv_data = output.getvalue()
    print(f'Generated dataset with 100 rows')
    return csv_data

What does this code do?

The from kfp import dsl and from kfp import compiler lines bring in the two KFP modules you need: dsl for defining components and pipelines, and compiler for converting the pipeline to YAML.

The load_data function generates 100 rows of synthetic data with three numeric features and a binary label. It writes everything to an in-memory CSV string so KFP can pass it between pods.

Why are the imports inside the function?

Each @dsl.component runs in its own Kubernetes pod with a fresh container. The pod only has access to packages available in the base_image (here, python:3.12-slim). Imports placed outside the function exist only on your Cloud Shell machine, not inside the pod.

That is why csv, io, and random are imported inside the function body. They ship with Python's standard library, so they are available in python:3.12-slim without any extra pip install.

  • Add the preprocess_data component below load_data in ml_pipeline.py:
@dsl.component(base_image='python:3.12-slim')
def preprocess_data(raw_csv: str) -> str:
    import csv
    import io

    reader = csv.DictReader(io.StringIO(raw_csv))
    rows = list(reader)

    features = ['feature1', 'feature2', 'feature3']
    mins = {}
    maxs = {}
    for feat in features:
        values = [float(row[feat]) for row in rows]
        mins[feat] = min(values)
        maxs[feat] = max(values)

    output = io.StringIO()
    writer = csv.DictWriter(output, fieldnames=features + ['label'])
    writer.writeheader()

    for row in rows:
        normalized = {}
        for feat in features:
            val = float(row[feat])
            range_val = maxs[feat] - mins[feat]
            normalized[feat] = round((val - mins[feat]) / range_val, 4) if range_val > 0 else 0
        normalized['label'] = row['label']
        writer.writerow(normalized)

    processed_csv = output.getvalue()
    print(f'Preprocessed {len(rows)} rows with min-max normalization')
    return processed_csv

What does preprocess_data do?

This component takes the raw CSV string from load_data and applies min-max normalization to each feature column. Every value gets rescaled to a 0-to-1 range. The label column passes through unchanged.

Notice the function signature: raw_csv: str tells KFP this component expects a string input. KFP uses these type annotations to wire data between pods automatically.

  • Save ml_pipeline.py.
  • Confirm there are no syntax errors by running:
python3 ml_pipeline.py

You should see no output and no errors. The file defines two components but has no pipeline definition or main block yet, so nothing executes.

Seeing a syntax error?

Check that preprocess_data is indented consistently. Every line inside the function must use four spaces, and the nested for loop body must use eight spaces. Also confirm you did not accidentally paste the second component inside the first function's body.

help me debug a syntax error in my KFP component file

Add the training, evaluation, and pipeline definition

With data loading and preprocessing in place, you need two more components: one to train a model and one to evaluate it. Then you will wire all four together into a pipeline and compile it.

  • Add the train_model component below preprocess_data in ml_pipeline.py:
@dsl.component(base_image='python:3.12-slim')
def train_model(processed_csv: str) -> float:
    import csv
    import io
    import random

    random.seed(42)
    reader = csv.DictReader(io.StringIO(processed_csv))
    rows = list(reader)

    split_idx = int(len(rows) * 0.8)
    train_data = rows[:split_idx]
    test_data = rows[split_idx:]

    threshold = 0.5
    correct = 0
    for row in test_data:
        avg_features = (float(row['feature1']) + float(row['feature2'])) / 2
        prediction = 1 if avg_features > threshold else 0
        if prediction == int(row['label']):
            correct += 1

    accuracy = round(correct / len(test_data), 4) if test_data else 0.0
    print(f'Trained on {len(train_data)} rows, tested on {len(test_data)} rows')
    print(f'Model accuracy: {accuracy}')
    return accuracy

What does train_model do?

This component splits the preprocessed data into 80% training and 20% test sets. It uses a simple threshold classifier: if the average of feature1 and feature2 exceeds 0.5, it predicts label 1.

The return type -> float tells KFP the output is a numeric accuracy value. KFP tracks this as a pipeline artifact that the next component can consume.

  • Add the evaluate_model component, the pipeline definition, and the main block below train_model in ml_pipeline.py:
@dsl.component(base_image='python:3.12-slim')
def evaluate_model(accuracy: float):
    print(f'Final model evaluation:')
    print(f'  Accuracy: {accuracy}')
    if accuracy >= 0.8:
        print('  Result: GOOD - Model meets production threshold')
    elif accuracy >= 0.6:
        print('  Result: ACCEPTABLE - Model needs improvement')
    else:
        print('  Result: POOR - Model requires retraining')


@dsl.pipeline(name='ml-training-pipeline')
def ml_pipeline():
    load_task = load_data()
    preprocess_task = preprocess_data(raw_csv=load_task.output)
    train_task = train_model(processed_csv=preprocess_task.output)
    evaluate_model(accuracy=train_task.output)


if __name__ == '__main__':
    compiler.Compiler().compile(ml_pipeline, 'ml_pipeline.yaml')
    print('Pipeline compiled to ml_pipeline.yaml')

What does this code do?

The evaluate_model component takes the accuracy float and prints a verdict: GOOD, ACCEPTABLE, or POOR. It is the final step in the pipeline.

The ml_pipeline function wires all four components together. Each line passes one component's .output as the next component's input, creating the DAG. KFP reads these connections to figure out which pods depend on which.

The if __name__ block compiles the pipeline to a YAML file that contains the full Argo Workflows spec. This is the same compilation step you used for the hello pipeline.

  • Save ml_pipeline.py.
  • Compile the pipeline by running:
python3 ml_pipeline.py

You should see Pipeline compiled to ml_pipeline.yaml printed in the terminal.

  • Confirm the compiled YAML file was created by running:
ls ml_pipeline.yaml

You should see ml_pipeline.yaml listed. This file contains the Argo Workflow spec with four tasks, container definitions, and the DAG wiring that connects each step's outputs to the next step's inputs.

Seeing an import error or compilation failure?

Make sure evaluate_model and the @dsl.pipeline function are at the top level of the file, not indented inside another function. Also confirm the if __name__ block is at the bottom with no indentation.

help me fix a compilation error in my KFP ml_pipeline.py file

✔️ Awesome, I've got everything!

Great. Double check you have saved ml_pipeline.py and that ml_pipeline.yaml exists in your Cloud Shell home directory.

ⓧ I'd like to double check the full code

Your ml_pipeline.py file is built from four code chunks. Scroll up and confirm each one is present in this order:

  • Chunk 1: from kfp import dsl and from kfp import compiler imports, followed by the load_data component.
  • Chunk 2: The preprocess_data component with min-max normalization.
  • Chunk 3: The train_model component with the threshold classifier.
  • Chunk 4: The evaluate_model component, the @dsl.pipeline wiring function, and the if __name__ compilation block.

Each chunk should be separated by two blank lines (between top-level definitions). Make sure no chunk is indented inside another function.

Upload and run the pipeline in the KFP UI

The compiled YAML is ready. Now you will upload it to the Kubeflow Pipelines dashboard and create a run to watch all four steps execute as Kubernetes pods.

  • First, download ml_pipeline.yaml from Cloud Shell to your local machine by running:
cloudshell download ml_pipeline.yaml

Your browser will prompt you to save the file. Save it somewhere you can find it easily, like your Desktop.

  • Make sure the port-forward to the KFP UI is still running. If it stopped, restart it by running this in a separate Cloud Shell terminal tab:
kubectl port-forward -n kubeflow svc/ml-pipeline-ui 8080:80
  • Open the KFP UI using Cloud Shell's Web Preview on port 8080.
  • In the left sidebar, click Pipelines.
  • Click + Upload pipeline in the top right.
  • Select Upload a file and choose the ml_pipeline.yaml file you just downloaded.
  • Enter ml-training-pipeline as the pipeline name.
  • Click Create.

You should see the pipeline details page with a DAG graph showing four nodes connected in sequence: load-data, preprocess-data, train-model, and evaluate-model.

  • Click Create run in the top right of the pipeline details page.
  • Leave the default experiment and parameters unchanged.
  • Click Start.

Before you watch the run, think about what you expect to happen. Do you think all four steps will succeed? What output will the training step produce?

Watch the DAG graph update in real time. Each node turns green as its pod completes. All four steps should finish within a few minutes.

  • Once all four nodes show green checkmarks, click the train-model node in the DAG.
  • Select the Logs tab or the Output tab to see the accuracy metric the component returned.

You should see the model accuracy value printed in the logs, along with the number of training and test rows.

What is happening under the hood?

Each green node in the DAG ran as a separate Kubernetes pod. You can verify this by running kubectl get pods -n kubeflow in your Cloud Shell terminal. You will see completed pods for each pipeline step.

KFP passed the CSV data from load_data to preprocess_data to train_model automatically. It serialized each component's output, stored it in Minio (the artifact store), and injected it as an input parameter into the next pod.

Pipeline step stuck in pending or failed?

If a step stays in a pending state, your cluster may be low on resources. Run kubectl get pods -n kubeflow to check for pods in a Pending or Error state. Make sure your three GKE nodes are still in Ready status with kubectl get nodes.

If a step failed with a red node, click it and check the Logs tab. The error message will point you to the exact line that failed in your component code.

help me debug a failed step in my Kubeflow pipeline run

Your four-step ML pipeline is running on Kubernetes. Each Python function you wrote became a containerized pod, and KFP orchestrated the entire flow as a DAG. Next up, take on the Secret Mission and add conditional branching to control which steps run based on the model's accuracy.

Secret mission

Add Conditional Branching

Your current ML pipeline always runs evaluation, no matter how the model performed. Real ML pipelines need decision points. Can you add a conditional branch that skips evaluation when accuracy is too low, and prints improvement suggestions instead?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely inside a KodeKloud GCP Playground session, which is free and auto-terminates after 3 hours. All resources are automatically cleaned up when your session ends, so there are no ongoing costs.

Resources you used:

  • GKE cluster kfp-cluster (3x e2-medium nodes in us-west1).
  • Kubeflow Pipelines v2.17.0 deployment in the kubeflow namespace.
  • Pipeline Python files and compiled YAML files in Cloud Shell (hello_pipeline.py, broken_pipeline.py, ml_pipeline.py, ml_pipeline_conditional.py, and their .yaml counterparts).

Keep everything running

No action needed. Choose this if you are still actively experimenting within your 3-hour playground window.

  • Your GKE cluster, Kubeflow Pipelines deployment, and all Cloud Shell files remain available until the session auto-terminates.
  • Try uploading different pipeline YAML files, tweaking component logic, or creating new runs from the KFP UI.
  • Keep an eye on the session timer in the KodeKloud dashboard. Once the 3 hours are up, everything is destroyed automatically.

Pause - I'll come back to this later

KodeKloud Playground sessions cannot be paused. Once a session starts, the 3-hour timer runs until it expires.

  • If your session ends before you finish, you can start a new playground session at any time. KodeKloud allows unlimited sessions.
  • Save any pipeline files you want to keep by copying them to your local machine before the session expires. You can download files from the Cloud Shell editor or copy the contents to a local text file.

Delete - I don't want to use this again

No manual cleanup is required. The KodeKloud Playground automatically destroys all resources when your session expires.

  • The GKE cluster kfp-cluster and all its nodes are deleted when the session ends.
  • The Kubeflow Pipelines deployment, including the metadata store and artifact storage, is destroyed along with the cluster.
  • All Cloud Shell files (.py scripts, compiled .yaml files, and the installed KFP SDK) are removed when the session ends.
  • If you want to end your session early, return to the KodeKloud dashboard and click End Lab to terminate the playground immediately.

Nice Work!

Nice Work!

Nice work! You just deployed a complete ML pipeline platform on Kubernetes and ran your own multi-step pipeline from scratch.

You've learned how to:

  • Created a GKE cluster and deployed Kubeflow Pipelines standalone using official kustomize manifests, then explored the architecture: API server, metadata store, artifact storage, Argo Workflows engine, and web UI.
  • Wrote KFP components using @dsl.component decorators that turn Python functions into containerized pipeline steps, and compiled pipelines to YAML to see how they map to Argo Workflow specs and Kubernetes pod definitions.
  • Built and ran a multi-step ML pipeline with four stages (load, preprocess, train, evaluate) that pass data artifacts between pods, with real-time execution tracking in the KFP UI.
  • Secret Mission: Added conditional branching using dsl.If and dsl.Else to control pipeline execution based on runtime metrics.

Ready to quiz yourself?