Kubeflow Pipelines

https://www.kubeflow.org/docs/components/pipelines/
Admin
Kubeflow Pipelines is a platform for building, deploying, and managing end-to-end machine learning...
Tokens:37,099
Snippets:470
Trust Score:10
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install kfp-kubernetes library

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/platform-specific-features

Install the required library to enable Kubernetes-specific functionality in your KFP environment.

```bash
pip install kfp[kubernetes]
```

--------------------------------

### Install KFP SDK

Source: https://www.kubeflow.org/docs/components/pipelines/getting-started

Install the Kubeflow Pipelines SDK version 2 using pip.

```bash
pip install kfp
```

--------------------------------

### Translate pip_index_urls to command line

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/lightweight-python-components

Example of how pip_index_urls arguments translate to the underlying pip install command.

```bash
pip install custom-ml-package==0.0.1 numpy==1.21.6 kfp==2 --index-url http://myprivaterepo.com/simple --trusted-host http://myprivaterepo.com/simple --extra-index-url http://pypi.org/simple --trusted-host http://pypi.org/simple
```

--------------------------------

### Example Provider Configuration for PipelineRoot

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/configure-object-store

Shows a specific provider configuration matching a defined PipelineRoot path.

```yaml
- bucketName: your-bucket
  keyPrefix: some/s3/path/b/a
  endpoint: https://s3.amazonaws.com
  region: us-east-2
  credentials:
    fromEnv: false
    secretRef:
      secretName: aws-s3-creds
      accessKeyKey: AWS_ACCESS_KEY_ID
      secretKeyKey: AWS_SECRET_ACCESS_KEY
```

--------------------------------

### Building containerized components

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/cli

Commands for installing dependencies and building container images for Python components.

```bash
pip install kfp[all]
```

```bash
kfp component build [OPTIONS] [COMPONENTS_DIRECTORY] [ARGS]...
```

```bash
kfp component build src/ --component-filepattern my_component --push-image
```

--------------------------------

### Check KFP CLI installation

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/cli

Verify that the KFP CLI is installed and accessible in the current environment.

```bash
kfp --version
```

--------------------------------

### Install dependencies with packages_to_install

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/lightweight-python-components

Specify external Python libraries to be installed at runtime.

```python
@dsl.component(packages_to_install=['numpy==1.21.6'])
def sin(val: float = 3.14) -> float:
    return np.sin(val).item()
```

--------------------------------

### Traditional Artifact Implementation

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/artifacts

Example of reading from an input artifact and writing to an output artifact using the traditional syntax.

```python
from kfp import dsl
from kfp.dsl import Dataset, Input, Model, Output

@dsl.component
def train_model(dataset: Input[Dataset], model: Output[Model]):
    with open(dataset.path) as f:
        dataset_lines = f.readlines()

    # train a model
    trained_model = ...
    
    trained_model.save(model.path)
    model.metadata['samples'] = len(dataset_lines)
```

--------------------------------

### Install Python packages at component runtime

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/build-advanced-pipeline

Use the packages_to_install argument within the @dsl.component decorator to specify dependencies.

```python
@dsl.component(packages_to_install=['pandas==1.3.5'])
```

--------------------------------

### Define and Execute an Advanced ML Pipeline

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/build-advanced-pipeline

This example defines components for dataset creation, normalization, and model training, then executes them within a pipeline using parallel processing.

```python
from typing import List

from kfp import client
from kfp import dsl
from kfp.dsl import Dataset
from kfp.dsl import Input
from kfp.dsl import Model
from kfp.dsl import Output


@dsl.component(packages_to_install=['pandas==1.3.5'])
def create_dataset(iris_dataset: Output[Dataset]):
    import pandas as pd

    csv_url = 'https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data'
    col_names = [
        'Sepal_Length', 'Sepal_Width', 'Petal_Length', 'Petal_Width', 'Labels'
    ]
    df = pd.read_csv(csv_url, names=col_names)

    with open(iris_dataset.path, 'w') as f:
        df.to_csv(f)


@dsl.component(packages_to_install=['pandas==1.3.5', 'scikit-learn==1.0.2'])
def normalize_dataset(
    input_iris_dataset: Input[Dataset],
    normalized_iris_dataset: Output[Dataset],
    standard_scaler: bool,
    min_max_scaler: bool,
):
    if standard_scaler is min_max_scaler:
        raise ValueError(
            'Exactly one of standard_scaler or min_max_scaler must be True.')

    import pandas as pd
    from sklearn.preprocessing import MinMaxScaler
    from sklearn.preprocessing import StandardScaler

    with open(input_iris_dataset.path) as f:
        df = pd.read_csv(f)
    labels = df.pop('Labels')

    if standard_scaler:
        scaler = StandardScaler()
    if min_max_scaler:
        scaler = MinMaxScaler()

    df = pd.DataFrame(scaler.fit_transform(df))
    df['Labels'] = labels
    with open(normalized_iris_dataset.path, 'w') as f:
        df.to_csv(f)


@dsl.component(packages_to_install=['pandas==1.3.5', 'scikit-learn==1.0.2'])
def train_model(
    normalized_iris_dataset: Input[Dataset],
    model: Output[Model],
    n_neighbors: int,
):
    import pickle

    import pandas as pd
    from sklearn.model_selection import train_test_split
    from sklearn.neighbors import KNeighborsClassifier

    with open(normalized_iris_dataset.path) as f:
        df = pd.read_csv(f)

    y = df.pop('Labels')
    X = df

    X_train, X_test, y_train, y_test = train_test_split(X, y, random_state=0)

    clf = KNeighborsClassifier(n_neighbors=n_neighbors)
    clf.fit(X_train, y_train)
    with open(model.path, 'wb') as f:
        pickle.dump(clf, f)


@dsl.pipeline(name='iris-training-pipeline')
def my_pipeline(
    standard_scaler: bool,
    min_max_scaler: bool,
    neighbors: List[int],
):
    create_dataset_task = create_dataset()

    normalize_dataset_task = normalize_dataset(
        input_iris_dataset=create_dataset_task.outputs['iris_dataset'],
        standard_scaler=True,
        min_max_scaler=False)

    with dsl.ParallelFor(neighbors) as n_neighbors:
        train_model(
            normalized_iris_dataset=normalize_dataset_task
            .outputs['normalized_iris_dataset'],
            n_neighbors=n_neighbors)


endpoint = '<KFP_UI_URL>'
kfp_client = client.Client(host=endpoint)
run = kfp_client.create_run_from_pipeline_func(
    my_pipeline,
    arguments={
        'min_max_scaler': True,
        'standard_scaler': False,
        'neighbors': [3, 6, 9]
    },
)
url = f'{endpoint}/#/runs/details/{run.run_id}'
print(url)
```

--------------------------------

### Example IR YAML Header

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/compile-a-pipeline

The generated IR YAML file contains metadata comments at the top describing the pipeline structure.

```yaml
# PIPELINE DEFINITION
# Name: my-pipeline
# Description: My ML pipeline.
# Inputs:
#    message: str
# Outputs:
#    Output: str
...
```

--------------------------------

### List pipeline runs via REST API

Source: https://www.kubeflow.org/docs/components/pipelines/reference/api/kubeflow-pipeline-api-spec

Example of a GET request to retrieve pipeline runs filtered by namespace.

```http
https://kubeflow.example.com/pipeline/apis/v2beta1/runs?namespace=team-1
```

--------------------------------

### Use component in a pipeline

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/containerized-python-components

Example of integrating the containerized component into a KFP pipeline definition.

```python
# pipeline.py
from kfp import compiler, dsl
from src.my_component import add

@dsl.pipeline
def addition_pipeline(x: int, y: int) -> int:
    task1 = add(a=x, b=y)
    task2 = add(a=task1.output, b=x)
    return task2.output

compiler.Compiler().compile(addition_pipeline, 'pipeline.yaml')
```

--------------------------------

### Generated Kubernetes Manifest Structure

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/compile-a-pipeline

Example of the resulting YAML structure containing Pipeline and PipelineVersion custom resources.

```yaml
apiVersion: pipelines.kubeflow.org/v2beta1
kind: Pipeline
metadata:
  name: hello-world-pipeline
  namespace: kubeflow-user-example-com
spec:
  displayName: Hello World Pipeline
---
apiVersion: pipelines.kubeflow.org/v2beta1
kind: PipelineVersion
metadata:
  name: hello-world-pipeline-v1
  namespace: kubeflow-user-example-com
spec:
  displayName: Hello World Pipeline v1
  pipelineName: hello-world-pipeline
  pipelineSpec:
    # Pipeline specification details
  platformSpec:
    # Platform specification details
```

--------------------------------

### Pythonic Artifact Implementation

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/artifacts

Example of constructing and returning an artifact object using the Pythonic syntax.

```python
from kfp import dsl
from kfp.dsl import Dataset, Model

@dsl.component
def train_model(dataset: Dataset) -> Model:
    with open(dataset.path) as f:
        dataset_lines = f.readlines()

    # train a model
    trained_model = ...

    model_artifact = Model(uri=dsl.get_uri(), metadata={'samples': len(dataset_lines)})
    trained_model.save(model_artifact.path)
    
    return model_artifact
```

--------------------------------

### Configure API Server Proxy Settings

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/server-config

Example Kubernetes deployment manifest for the ml-pipeline API server with HTTP_PROXY, HTTPS_PROXY, and NO_PROXY environment variables defined.

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  labels:
    app: ml-pipeline
    application-crd-id: kubeflow-pipelines
  name: ml-pipeline
  namespace: kubeflow
spec:
  selector:
    matchLabels:
      app: ml-pipeline
      application-crd-id: kubeflow-pipelines
  template:
    metadata:
      annotations:
        cluster-autoscaler.kubernetes.io/safe-to-evict: "true"
      labels:
        app: ml-pipeline
        application-crd-id: kubeflow-pipelines
    spec:
      containers:
      - env:
        - name: HTTP_PROXY
          value: http://squid.squid.svc.cluster.local:3128
        - name: HTTPS_PROXY
          value: http://squid.squid.svc.cluster.local:3128
        - name: NO_PROXY
          value: localhost,127.0.0.1,.svc.cluster.local,kubernetes.default.svc,metadata-grpc-service,0,1,2,3,4,5,6,7,8,9
```

--------------------------------

### Deploy Kubeflow Pipelines in Kubernetes Native API Mode

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/installation

Deploys KFP using Kubernetes Custom Resources instead of external database storage. Requires cert-manager to be installed.

```bash
export PIPELINE_VERSION=2.17.0

kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/cluster-scoped-resources?ref=$PIPELINE_VERSION"
kubectl wait --for condition=established --timeout=60s crd/applications.app.k8s.io
kubectl apply -f https://github.com/cert-manager/cert-manager/releases/download/v1.18.2/cert-manager.yaml
kubectl wait --for=condition=Ready pod -l app.kubernetes.io/instance=cert-manager -n cert-manager --timeout=300s
kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/env/cert-manager/platform-agnostic-k8s-native?ref=$PIPELINE_VERSION"
# Alternatively, for multi-user environments with multiple teams or users requiring isolation and RBAC controls on who can access which pipelines (still not production-ready like the community distribution), you can use the multi-user Kubernetes native mode (requires Istio to be installed, so we strongly recommend using the community distribution instead):
# kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/env/cert-manager/platform-agnostic-multi-user-k8s-native?ref=$PIPELINE_VERSION"
```

--------------------------------

### Authenticate and Initialize KFP Client

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/connect-api

Demonstrates how to instantiate the KFPClientManager with credentials and retrieve an authenticated kfp.Client instance.

```python
# initialize a KFPClientManager
kfp_client_manager = KFPClientManager(
    api_url="http://localhost:8080/pipeline",
    skip_tls_verify=True,

    dex_username="user@example.com",
    dex_password="12341234",

    # can be 'ldap' or 'local' depending on your Dex configuration
dex_auth_type="local",
)

# get a newly authenticated KFP client
# TIP: long-lived sessions might need to get a new client when their session expires
kfp_client = kfp_client_manager.create_kfp_client()

# test the client by listing experiments
experiments = kfp_client.list_experiments(namespace="my-profile")
print(experiments)
```

--------------------------------

### Importing Artifacts to Workspace in KFP

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/importer-component

Demonstrates configuring the importer component with download_to_workspace=True and setting up a pipeline workspace configuration.

```python
from kfp import dsl
from kfp.dsl import Output, importer

@dsl.component()
def train(dataset: dsl.Input[dsl.Dataset]):
    """Dummy Training step."""
    with open(dataset.path, encoding="utf-8") as f:
        data = f.read()

@dsl.component()
def write_file_artifact(out_ds: Output[dsl.Dataset]):
    import os
    os.makedirs(os.path.dirname(out_ds.path), exist_ok=True)
    with open(out_ds.path, "w", encoding="utf-8") as f:
        f.write("Hello from producer file\n")

@dsl.pipeline()
def import_stage(file_uri: str):
    """Nested stage that imports by URI and runs consumers."""
    importer1 = importer(
        artifact_uri=file_uri,
        artifact_class=dsl.Dataset,
        download_to_workspace=True,
    )

    return train(dataset=importer1.output)

@dsl.pipeline(
    pipeline_config=dsl.PipelineConfig(
        workspace=dsl.WorkspaceConfig(
            size='1Gi',
            kubernetes=dsl.KubernetesWorkspaceConfig(
                pvcSpecPatch={'storageClassName': 'standard','accessModes': ['ReadWriteOnce']}
            ),
        ),
    ),
)
def pipeline_with_importer_workspace():
    # Produce a file artifact and compute its runtime URI
    file_writer = write_file_artifact()

    # Import and consume inside a nested sub-pipeline
    stage = import_stage(file_uri=file_writer.outputs["out_ds"].uri)
```

--------------------------------

### Instantiate KFP Client

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/run-a-pipeline

Initialize the KFP client to interact with the API. Authentication may be required depending on the environment.

```python
import kfp

# TIP: you may need to authenticate with the KFP instance
kfp_client = kfp.Client()
```

--------------------------------

### GET /apis/v2beta1/runs

Source: https://www.kubeflow.org/docs/components/pipelines/reference/api/kubeflow-pipeline-api-spec

Lists pipeline runs within a specified namespace.

```APIDOC
## GET /apis/v2beta1/runs

### Description
Retrieves a list of pipeline runs filtered by the specified namespace.

### Method
GET

### Endpoint
/apis/v2beta1/runs

### Parameters
#### Query Parameters
- **namespace** (string) - Required - The namespace to filter pipeline runs by.
```

--------------------------------

### Execute Components and Pipelines Locally

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/execute-kfp-pipelines-locally

Initializes the local environment with DockerRunner and demonstrates running a single component and a full pipeline.

```python
from kfp import local
from kfp import dsl

local.init(runner=local.DockerRunner())

@dsl.component
def add(a: int, b: int) -> int:
    return a + b

# run a single component
task = add(a=1, b=2)
assert task.output == 3

# or run it in a pipeline
@dsl.pipeline
def math_pipeline(x: int, y: int, z: int) -> int:
    t1 = add(a=x, b=y)
    t2 = add(a=t1.output, b=z)
    return t2.output

pipeline_task = math_pipeline(x=1, y=2, z=3)
assert pipeline_task.output == 6
```

--------------------------------

### Configure custom pip indices

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/lightweight-python-components

Use pip_index_urls to install packages from private repositories.

```python
@dsl.component(packages_to_install=['custom-ml-package==0.0.1', 'numpy==1.21.6'],
               pip_index_urls=['http://myprivaterepo.com/simple', 'http://pypi.org/simple'],
)
def comp():
    from custom_ml_package import model_trainer
    import numpy as np
    ...
```

--------------------------------

### Configure pipeline metadata with the decorator

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/compose-components-into-pipelines

Shows how to provide optional metadata like name, description, and pipeline_root to the @dsl.pipeline decorator.

```python
@dsl.pipeline(name='pythagorean-theorem-pipeline',
              description='Solve for the length of a hypotenuse of a triangle with sides length `a` and `b`.',
              pipeline_root='gs://my-pipelines-bucket',
              display_name='Pythagorean pipeline.')
def pythagorean(a: float, b: float) -> float:
    ...
```

--------------------------------

### Accessing command help

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/cli

Use the --help flag to view documentation for specific CLI commands.

```bash
kfp run --help
```

--------------------------------

### Resource management syntax

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/cli

General pattern for performing CRUD actions on KFP resources.

```bash
kfp <resource_name> <action>
```

--------------------------------

### Initialize kfp.Client from inside a cluster

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/connect-api

Uses the default ServiceAccount token path and API host to authenticate the client when running within a Kubernetes cluster.

```python
import kfp

# by default, when run from inside a Kubernetes cluster:
#  - the token is read from the `KF_PIPELINES_SA_TOKEN_PATH` path
#  - the host is set to `http://ml-pipeline-ui.kubeflow.svc.cluster.local`
kfp_client = kfp.Client()

# test the client by listing experiments
experiments = kfp_client.list_experiments(namespace="my-profile")
print(experiments)
```

--------------------------------

### Custom Pipeline Root Path

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/configure-object-store

Example of modifying the defaultPipelineRoot to store artifacts in a specific path within the bucket.

```yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: kfp-launcher
  namespace: user-namespace
data:
  defaultPipelineRoot: "minio://mlpipeline/some/other/path"
```

--------------------------------

### Build component image

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/containerized-python-components

CLI command to build the container image for the component.

```bash
kfp component build src/ --component-filepattern my_component.py --no-push-image
```

--------------------------------

### Deploy Kubeflow Pipelines development flavor

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/installation

Use this for non-production experiments. Requires kubectl and kustomize.

```bash
export PIPELINE_VERSION=2.17.0

kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/cluster-scoped-resources?ref=$PIPELINE_VERSION"
kubectl wait --for condition=established --timeout=60s crd/applications.app.k8s.io
kubectl apply -k "github.com/kubeflow/pipelines/manifests/kustomize/env/dev?ref=$PIPELINE_VERSION"
```

--------------------------------

### Identify Typecasting Incompatibility

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/migration

Example of a pipeline that fails in KFP SDK v2 due to strict type enforcement where a string is passed to a float parameter.

```python
from kfp.v2 import compiler
from kfp.v2 import dsl
from kfp import components


@dsl.component
def train(
    number_of_epochs: int,
    learning_rate: float,
):
    print(f"number_of_epochs={number_of_epochs}")
    print(f"learning_rate={learning_rate}")


def training_pipeline(number_of_epochs: int = 1):
    train(
        number_of_epochs=number_of_epochs,
        learning_rate="0.1",  # string cannot be passed to float parameter using KFP SDK v2
    )
```

--------------------------------

### Define a basic notebook component

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/notebook-component

Use the notebook_path argument to specify the target notebook file for execution within the component.

```python
@dsl.notebook_component(
    notebook_path="train.ipynb",)
def evaluate_model(dataset_uri: str, model: dsl.Output[dsl.Model]):
    dsl.run_notebook(
        dataset_uri=dataset_uri,
        output_model_path=model.path,
    )
```

--------------------------------

### Update Import Namespaces

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/migration

Transition from the v2-specific namespace to the primary KFP namespace.

```python
from kfp.v2 import dsl
from kfp.v2 import compiler

@dsl.pipeline(name='my-pipeline')
def pipeline():
  ...

compiler.Compiler().compile(...)
```

```python
from kfp import dsl
from kfp import compiler

@dsl.pipeline(name='my-pipeline')
def pipeline():
  ...

compiler.Compiler().compile(...)
```

--------------------------------

### Configure Pipeline Run Workspace and Component Data Sharing

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/artifacts

Demonstrates defining a workspace in a pipeline configuration and using dsl.WORKSPACE_PATH_PLACEHOLDER to access the shared volume within components.

```python
from kfp import dsl
from kfp import compiler


@dsl.component()
def clone_repo(workspacePath: str, repo: str) -> str:
    import os
    import subprocess

    clone_path = os.path.join(workspacePath, "repo")
    subprocess.call(["git", "clone", repo, clone_path])

    return clone_path


@dsl.component()
def process_data(repo_path: str):
    print("Processing the data at " + repo_path)


@dsl.component()
def train(model: dsl.Model, trained_model: dsl.Output[dsl.Model]):
    with open(model.path, "r") as model_file:
        print("Training the model")

    # Upload the model to S3 and register it in MLMD
    trained_model.set_path(model.path) # or trained_model.custom_path = model.path
    
@dsl.pipeline(
    name="my-pipeline",
    pipeline_config=dsl.PipelineConfig(
        workspace=dsl.WorkspaceConfig(
            size="250GB",
            kubernetes=dsl.KubernetesWorkspaceConfig(
                pvcSpecPatch={
                    "storageClassName": "super-fast-storage",
                    "accessModes": ["ReadWriteMany"],
                }
            ),
        ),
    )
def pipeline(repo: str, model_uri: str):
    clone_repo_task = clone_repo(
        workspacePath=dsl.WORKSPACE_PATH_PLACEHOLDER, repo=repo,
    )
    process_data_task = process_data(repo_path=clone_repo_task.output)
    import_base_model_task = dsl.importer(
        artifact_class=dsl.Model, artifact_uri=model_uri, download_to_workspace=True,
    )
    train(model=import_base_model_task.output).after(process_data_task)
```

--------------------------------

### Build component artifacts with KFP CLI

Source: https://www.kubeflow.org/docs/components/pipelines/concepts/component

Generates Dockerfile and runtime-dependencies.txt for a component defined in a Python file without building the image.

```bash
# build Dockerfile and runtime-dependencies.txt
kfp component build --component-filepattern the_component.py --no-build-image --platform linux/amd64 .
```

--------------------------------

### Connect to KFP from Outside the Cluster

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/connect-api

Initiate port-forwarding to the service and connect the client using localhost.

```bash
# change `--namespace` if you deployed Kubeflow Pipelines into a different namespace
kubectl port-forward --namespace kubeflow svc/ml-pipeline-ui 3000:80
```

```python
import kfp

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

print(client.list_experiments())
```

--------------------------------

### Manage and Verify MLflow Plugin Deployment

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/mlflow-plugin

Commands to restart the KFP API server and monitor logs for configuration verification.

```bash
kubectl rollout restart deployment/ml-pipeline -n kubeflow
```

```bash
kubectl logs -n kubeflow deployment/ml-pipeline | grep -i mlflow
```

--------------------------------

### Submit Pipeline for Execution

Source: https://www.kubeflow.org/docs/components/pipelines/getting-started

Use the KFP Client to submit the compiled YAML file to a KFP-conformant backend.

```python
from kfp.client import Client

client = Client(host='<MY-KFP-ENDPOINT>')
run = client.create_run_from_pipeline_package(
    'pipeline.yaml',
    arguments={
        'recipient': 'World',
    },
)
```

--------------------------------

### Define a training component with input and output artifacts

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/artifacts

Demonstrates using Input and Output artifact types to read a dataset and save a trained model.

```python
from kfp.dsl import Input, Output, Dataset, Model

@dsl.component
def training_component(dataset: Input[Dataset], model: Output[Model]):
    """Trains an output Model on an input Dataset."""
    with open(dataset.path) as f:
        contents = f.read()

    # ... train tf_model model on contents of dataset ...

    tf_model.save(model.path)
    model.metadata['framework'] = 'tensorflow'
```

--------------------------------

### Create a notebook component

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/notebook-component

Uses the @dsl.notebook_component decorator to embed a Jupyter notebook and execute it at runtime using nbclient.

```python
@dsl.notebook_component(
    notebook_path="train.ipynb",
    packages_to_install=["pandas", "scikit-learn", "nbclient"],
)
def evaluate_model(dataset_uri: str, model: dsl.Output[dsl.Model]):
    dsl.run_notebook(
        dataset_uri=dataset_uri,
        output_model_path=model.path,
    )
```

--------------------------------

### KFP CLI general syntax

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/cli

The standard structure for executing KFP CLI commands.

```bash
kfp [OPTIONS] COMMAND [ARGS]...
```

```bash
kfp --endpoint http://my_kfp_endpoint.com run list
```

--------------------------------

### Submit IR YAML via SDK

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/run-a-pipeline

Execute a pre-compiled pipeline package using the create_run_from_pipeline_package method.

```python
#from kfp import compiler, dsl
#
#@dsl.component
#def add(a: float, b: float) -> float:
#   return a + b
#
#@dsl.pipeline(name="Add two Numbers")
#def add_pipeline(a: float, b: float):
#   add_task = add(a=a, b=b)
#
#compiler.Compiler().compile(
#    add_pipeline, 
#    package_path="./add-pipeline.yaml"
#)

kfp_client.create_run_from_pipeline_package(
    "./add-pipeline.yaml", 
    arguments={
        "a": 1,
        "b": 2,
    }
)
```

--------------------------------

### Implement input parameters in components and pipelines

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/parameters

Demonstrates passing various data types including strings, integers, floats, booleans, dictionaries, and lists to Python and container components.

```python
from typing import Dict, List
from kfp import dsl

@dsl.component
def python_comp(
    string: str = 'hello',
    integer: int = 1,
    floating_pt: float = 0.1,
    boolean: bool = True,
    dictionary: Dict = {'key': 'value'},
    array: List = [1, 2, 3],
):
    print(string)
    print(integer)
    print(floating_pt)
    print(boolean)
    print(dictionary)
    print(array)


@dsl.container_component
def container_comp(
    string: str = 'hello',
    integer: int = 1,
    floating_pt: float = 0.1,
    boolean: bool = True,
    dictionary: Dict = {'key': 'value'},
    array: List = [1, 2, 3],
):
    return dsl.ContainerSpec(
        image='alpine',
        command=['sh', '-c', """echo $0 $1 $2 $3 $4 $5 $6"""],
        args=[
            string,
            integer,
            floating_pt,
            boolean,
            dictionary,
            array,
        ])

@dsl.pipeline
def my_pipeline(
    string: str = 'Hey!',
    integer: int = 100,
    floating_pt: float = 0.1,
    boolean: bool = False,
    dictionary: Dict = {'key': 'value'},
    array: List = [1, 2, 3],
):
    python_comp(
        string='howdy',
        integer=integer,
        array=[4, 5, 6],
    )
    container_comp(
        string=string,
        integer=20,
        dictionary={'other key': 'other val'},
        boolean=boolean,
    )
```

--------------------------------

### Define a basic pipeline with components

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/compose-components-into-pipelines

Demonstrates the structure of a pipeline function using arithmetic components to implement the Pythagorean theorem.

```python
from kfp import dsl

@dsl.component
def square(x: float) -> float:
    return x ** 2

@dsl.component
def add(x: float, y: float) -> float:
    return x + y

@dsl.component
def square_root(x: float) -> float:
    return x ** .5

@dsl.pipeline
def pythagorean(a: float, b: float) -> float:
    a_sq_task = square(x=a)
    b_sq_task = square(x=b)
    sum_task = add(x=a_sq_task.output, y=b_sq_task.output)
    return square_root(x=sum_task.output).output
```

--------------------------------

### Define a simple KFP pipeline

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/integrations/mlflow/pipeline-structure-in-mlflow

A basic pipeline definition using dsl.component and dsl.pipeline decorators.

```python
from kfp import dsl


@dsl.component()
def print_op(text: str):
    print(text)

@dsl.pipeline()
def pipeline():
    task = print_op(text='hello world!')
```

--------------------------------

### Initialize DockerRunner

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/execute-kfp-pipelines-locally

Configures the local environment to use DockerRunner for pipeline execution.

```python
from kfp import local

local.init(runner=local.DockerRunner())
```

--------------------------------

### Implement basic cleanup with dsl.ExitHandler

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/control-flow

Use the dsl.ExitHandler context manager to ensure a cleanup task runs after the tasks within its scope finish.

```python
from kfp import dsl
from kfp.dsl import Dataset

#@dsl.component
#def clean_up_resources():
#    ...

#@dsl.component
#def create_datasets():
#    ...

#@dsl.component
#def train_and_save_models(dataset: Dataset):
#    ...

@dsl.pipeline
def my_pipeline():
    clean_up_task = clean_up_resources()
    with dsl.ExitHandler(exit_task=clean_up_task):
        dataset_task = create_datasets()
        train_task = train_and_save_models(dataset=dataset_task.output)
```

--------------------------------

### Migrate component creation to @dsl.component

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/migration

Replaces the deprecated create_component_from_func and func_to_container_op functions with the @dsl.component decorator.

```python
from kfp.components import create_component_from_func
from kfp.components import func_to_container_op

@create_component_from_func
def component1(...):
    ...

def component2(...):
    ...

component2 = create_component_from_func(component2)

@func_to_container_op
def component3(...):
    ...

@dsl.pipeline(name='my-pipeline')
def pipeline():
    component1(...)
    component2(...)
    component3(...)
```

```python
from kfp import dsl

@dsl.component
def component1(...):
    ...

@dsl.component
def component2(...):
    ...

@dsl.component
def component3(...):
    ...

@dsl.pipeline(name='my-pipeline')
def pipeline():
    component1(...)
    component2(...)
    component3(...)
```

--------------------------------

### Compile Pipeline to Kubernetes Manifests

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/compile-a-pipeline

Use the Compiler class with kubernetes_manifest_format set to True to generate native Kubernetes resources.

```python
import kfp
from kfp import dsl
from kfp.compiler.compiler_utils import KubernetesManifestOptions

@dsl.component
def hello_world(name: str = "World") -> str:    
    print(f"Hello, {name}!")
    return f"Hello, {name}!"

@dsl.pipeline(
    name="hello-world-pipeline",
    description="A simple hello world pipeline"
)
def hello_world_pipeline(name: str = "Kubernetes Native") -> str:
    hello_task = hello_world(name=name)
    return hello_task.output

# Compile directly to Kubernetes manifests
kfp.compiler.Compiler().compile(
    pipeline_func=hello_world_pipeline,
    package_path="hello-world-pipeline-k8s.yaml",
    kubernetes_manifest_format=True,
    kubernetes_manifest_options=KubernetesManifestOptions(
        pipeline_name="hello-world-pipeline",
        pipeline_display_name="Hello World Pipeline",
        pipeline_version_name="hello-world-pipeline-v1",
        pipeline_version_display_name="Hello World Pipeline v1",
        namespace="kubeflow-user-example-com",
        include_pipeline_manifest=True
    )
)
```

--------------------------------

### Configure GCS static credentials via SDK

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/configure-object-store

Set the application credentials path and mount the GCS credentials JSON file as a volume.

```python
# Specify the default APP Credential path
your_task.set_env_variable(name='GOOGLE_APPLICATION_CREDENTIALS', value='/gcloud/credentials.json')
# Mount the GCS Credentials JSON
kubernetes.use_secret_as_volume(your_task, secret_name='gcs-secret', mount_path='/gcloud')
```

--------------------------------

### Integrate a container component into a pipeline

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/container-components

Demonstrates how to use a container component within a pipeline definition and compile it.

```python
from kfp import dsl
from kfp import compiler

@dsl.pipeline
def hello_pipeline(person_to_greet: str) -> str:
    # greeting argument is provided automatically at runtime!
    hello_task = say_hello(name=person_to_greet)
    return hello_task.outputs['greeting']

compiler.Compiler().compile(hello_pipeline, 'pipeline.yaml')
```

--------------------------------

### Define a component with parameter inputs

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/parameters

Use Python type annotations to define input parameters and default values for a component.

```python
from kfp import dsl

@dsl.component
def join_words(word: str, count: int = 10) -> str:
    return ' '.join(word for _ in range(count))
```

--------------------------------

### Define producer and consumer components

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/platform-specific-features

Create components that perform file operations on a specific mount path, which will later be mapped to a PVC.

```python
from kfp import dsl

@dsl.component
def producer() -> str:
    with open('/data/file.txt', 'w') as file:
        file.write('Hello world')
    with open('/data/file.txt', 'r') as file:
        content = file.read()
    print(content)
    return content

@dsl.component
def consumer() -> str:
    with open('/data/file.txt', 'r') as file:
        content = file.read()
    print(content)
    return content
```

--------------------------------

### Define a basic component

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/containerized-python-components

Initial definition of a lightweight Python component using the dsl.component decorator.

```python
from kfp import dsl

@dsl.component
def add(a: int, b: int) -> int:
    return a + b
```

--------------------------------

### Connect to KFP from Inside the Cluster

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/connect-api

Use these snippets to connect to the Pipelines SDK when running within the same or a different Kubernetes namespace.

```python
import kfp

client = kfp.Client(host="http://ml-pipeline-ui:80")

print(client.list_experiments())
```

```python
import kfp

# the namespace in which you deployed Kubeflow Pipelines
namespace = "kubeflow" 

client = kfp.Client(host=f"http://ml-pipeline-ui.{namespace}")

print(client.list_experiments())
```

--------------------------------

### Define component docstrings with KFP style

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/additional-functionality

Use the KFP docstring style to document component inputs and outputs for automatic IR YAML generation.

```python
@dsl.component
def join_datasets(
    dataset_a: Input[Dataset],
    dataset_b: Input[Dataset],
    out_dataset: Output[Dataset],
) -> str:
    """Concatenates two datasets.

    Args:
        dataset_a: First dataset.
        dataset_b: Second dataset.

    Returns:
        out_dataset: The concatenated dataset.
        Output: The concatenated string.
    """
    ...
```

--------------------------------

### Consume input by file

Source: https://www.kubeflow.org/docs/components/pipelines/reference/component-spec

Use {inputPath} to provide the local file path where input data is stored.

```yaml
command: [program.py, --train-set, {inputPath: training_data}]
```

```python
task2 = component1(training_data=some_task1.outputs['some_data'])
```

```bash
program.py --train-set /inputs/train_data/data
```

--------------------------------

### Expose Kubeflow Pipelines API via Port-Forwarding

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/connect-api

Use this command to make the Istio ingress gateway accessible on your local machine. Ensure the service name and ports match your specific cluster deployment.

```bash
# TIP: svc/istio-ingressgateway may be called something else, 
#      or use different ports in your distribution
kubectl port-forward --namespace istio-system svc/istio-ingressgateway 8080:80
```

--------------------------------

### Define a pipeline with artifact inputs and outputs

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/artifacts

Shows a pipeline that accepts a Dataset and returns a Model, utilizing task outputs for artifact passing.

```python
from kfp import dsl
from kfp.dsl import Dataset, Model

@dsl.pipeline
def augment_and_train(dataset: Dataset) -> Model:
    augment_task = augment_dataset(dataset=dataset)
    return train_model(dataset=augment_task.output).output
```

--------------------------------

### Define YAML-based components

Source: https://www.kubeflow.org/docs/components/pipelines/concepts/component

Create a YAML file to define a component's metadata, inputs, outputs, and container implementation. This approach is useful for running system commands or binaries directly.

```yaml
# my_component.yaml file
name: my-component
description: "Component that outputs \"<string prefix>...<num>\""

inputs:
- {name: string prefix, type: String}
- {name: num, type: Integer}

outputs: []

implementation:
  container:
    image: python:3.12-slim-bookworm
    args:
    - echo
    - {inputValue: string prefix}
    - ...
    - {inputValue: num}
```

--------------------------------

### Global Caching Configuration for Compiler

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/caching

Demonstrates how the global environment variable affects caching behavior during compilation.

```python
@dsl.pipeline(name='my-pipeline')
def my_pipeline():
    task_1 = create_dataset()
    task_2 = create_dataset()
    task_1.set_caching_options(False)

Compiler().compile(
    pipeline_func=my_pipeline,
    package_path='my_pipeline.yaml',

)
```

```bash
KFP_DISABLE_EXECUTION_CACHING_BY_DEFAULT=true \
python my_pipeline.py
```

--------------------------------

### Create CA Certificate ConfigMap

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/mlflow-plugin

Create a Kubernetes ConfigMap to store the CA certificate for the MLflow server.

```bash
kubectl create configmap mlflow-ca-cert \
  --from-file=ca.crt=/path/to/your/ca-certificate.crt \
  -n kubeflow
```

--------------------------------

### Load a component from a local file

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/load-and-share-components

Use load_component_from_file to import a component definition stored in a local YAML file.

```python
from kfp import components

loaded_comp = components.load_component_from_file('component.yaml')

@dsl.pipeline
def my_pipeline():
    loaded_comp()
```

--------------------------------

### Handle Artifacts in Local Execution

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/execute-kfp-pipelines-locally

Demonstrates using SubprocessRunner to create, write, and read artifacts within a component.

```python
from kfp import local
from kfp import dsl
from kfp.dsl import Output, Artifact
import json

local.init(runner=local.SubprocessRunner())

@dsl.component
def add(a: int, b: int, out_artifact: Output[Artifact]):
    import json

    result = json.dumps(a + b)

    with open(out_artifact.path, 'w') as f:
        f.write(result)

    out_artifact.metadata['operation'] = 'addition'


task = add(a=1, b=2)

# can read artifact contents
with open(task.outputs['out_artifact'].path) as f:
    contents = f.read()

assert json.loads(contents) == 3
assert task.outputs['out_artifact'].metadata['operation'] == 'addition'
```

--------------------------------

### Configure S3 Authentication via KFP SDK

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/data-handling/pipeline-root

Use the add_op_transformer method to apply AWS credentials to pipeline operations.

```python
dsl.get_pipeline_conf().add_op_transformer(aws.use_aws_secret('xxx', ‘xxx’, ‘xxx’))
```

--------------------------------

### Inject token volume using PodDefault

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/core-functions/connect-api

Use a PodDefault resource to automatically inject the required token volume and environment variables into pods within a specific namespace.

```yaml
apiVersion: kubeflow.org/v1alpha1
kind: PodDefault
metadata:
  name: access-ml-pipeline
  namespace: "<YOUR_USER_PROFILE_NAMESPACE>"
spec:
  desc: Allow access to Kubeflow Pipelines
  selector:
    matchLabels:
      access-ml-pipeline: "true"
  env:
    - ## this environment variable is automatically read by `kfp.Client()`
      ## this is the default value, but we show it here for clarity
      name: KF_PIPELINES_SA_TOKEN_PATH
      value: /var/run/secrets/kubeflow/pipelines/token
  volumes:
    - name: volume-kf-pipeline-token
      projected:
        sources:
          - serviceAccountToken:
              path: token
              expirationSeconds: 7200
              ## defined by the `TOKEN_REVIEW_AUDIENCE` environment variable on the `ml-pipeline` deployment
              audience: pipelines.kubeflow.org      
  volumeMounts:
    - mountPath: /var/run/secrets/kubeflow/pipelines
      name: volume-kf-pipeline-token
      readOnly: true
```

--------------------------------

### Access the Kubeflow Pipelines UI

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/installation

Use port-forwarding to access the KFP UI locally after deployment.

```bash
kubectl port-forward -n kubeflow svc/ml-pipeline-ui 8080:80
```

--------------------------------

### Define a basic Container Component

Source: https://www.kubeflow.org/docs/components/pipelines/user-guides/components/container-components

Use the dsl.container_component decorator and return a dsl.ContainerSpec to define a component that executes a specific command in a container.

```python
from kfp import dsl

@dsl.container_component
def say_hello():
    return dsl.ContainerSpec(image='alpine', command=['echo'], args=['Hello'])
```

--------------------------------

### Configure API Server with Static Credentials

Source: https://www.kubeflow.org/docs/components/pipelines/operator-guides/configure-object-store

Add these environment variables to the ml-pipeline deployment to connect to an S3-compliant bucket using static credentials stored in a Kubernetes secret.

```yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: ml-pipeline
  namespace: kubeflow
spec:
  ...
  template:
    ...
    spec:
      containers:
      - name: ml-pipeline-api-server
        serviceAccountName: "ml-pipeline"
        env:
          ...
          - name: OBJECTSTORECONFIG_HOST
            value: "your-bucket" # e.g. s3.amazonaws.com
          - name: OBJECTSTORECONFIG_PORT
            value: "port" # e.g. 443   
          - name: OBJECTSTORECONFIG_REGION
            value: "region" # e.g. us-east-1
            # true if object store is on a secure connection 
          - name: OBJECTSTORECONFIG_SECURE
            value: "true"        
            # These env vars reference the values from a Kubernetes secret
            # this requires deploying the secret ahead of time, and filling out the
            # following values accordingly.
          - name: OBJECTSTORECONFIG_ACCESSKEY
            valueFrom:
              secretKeyRef:
                key: "some-key-1"
                name: "secret-name"
          - name: OBJECTSTORECONFIG_SECRETACCESSKEY
            valueFrom:
              secretKeyRef:
                key: "some-key-2"
                name: "secret-name"
```
Tokens:
Raw