Posts Tagged

Machine Learning Workflows

Mastering Advanced Pipeline Design: Conditional Execution and Loops in Kubeflow

April 20, 2023 Kubeflow Pipelines Loops

Mastering Advanced Pipeline Design: Conditional Execution and Loops in Kubeflow

Kubeflow Pipelines provide a powerful platform for building, deploying, and managing machine learning workflows. To create more complex and dynamic pipelines, you may need to use conditional execution and loops. In this tutorial, we will guide you through the process of implementing conditional execution and loops in Kubeflow Pipelines using Python.

Step 1: Define a Conditional Execution Function

To demonstrate conditional execution in Kubeflow Pipelines, we will create a simple pipeline that processes input data depending on a condition. First, let’s define a Python function for the conditional execution:

from typing import NamedTuple
from kfp.components import create_component_from_func

def process_data_conditional(input_data: str, condition: str) -> NamedTuple("Outputs", [("output_data", str)]):
    import json
    from collections import namedtuple
    if condition == "uppercase":
        output_data = input_data.upper()
    elif condition == "lowercase":
        output_data = input_data.lower()
    else:
        output_data = input_data
    Outputs = namedtuple("Outputs", ["output_data"])
    return Outputs(output_data)
process_data_conditional_component = create_component_from_func(process_data_conditional, output_component_file="process_data_conditional_component.yaml")

This function takes an input string and a condition as arguments. Depending on the condition, the input data will be converted to uppercase, lowercase, or remain unchanged.

Step 2: Implement the Pipeline with Conditional Execution

Now, let’s create a pipeline that uses the process_data_conditional function:

import kfp
from kfp import dsl

@dsl.pipeline(
    name="Conditional Execution Pipeline",
    description="A pipeline that demonstrates conditional execution."
)
def conditional_pipeline(input_data: str = "Hello, Kubeflow!", condition: str = "uppercase"):
    process_data = process_data_conditional_component(input_data, condition)
if __name__ == "__main__":
    kfp.compiler.Compiler().compile(conditional_pipeline, "conditional_pipeline.yaml")

In this pipeline, the process_data_conditional function is called with the input data and condition provided as arguments.

Step 3: Upload and Run the Pipeline with Different Conditions

Access the Kubeflow Pipelines dashboard by navigating to the URL provided during the setup process.
Click on the “Pipelines” tab in the left-hand sidebar.
Click the “Upload pipeline” button in the upper right corner.
In the “Upload pipeline” dialog, click “Browse” and select the conditional_pipeline.yaml file generated in the previous step.
Click “Upload” to upload the pipeline to the Kubeflow platform.
Once the pipeline is uploaded, click on its name to open the pipeline details page.
Click the “Create run” button to start a new run of the pipeline.
On the “Create run” page, you can give your run a name and choose a pipeline version. Set the “input_data” and “condition” arguments to test different conditions (e.g., “uppercase”, “lowercase”, or “unchanged”).
Click “Start” to begin the pipeline run.

Step 4: Add a Loop to the Pipeline

To demonstrate how to add loops in Kubeflow Pipelines, we will modify our pipeline to process a list of input data and conditions. First, let’s update the conditional_pipeline function:

import kfp
from kfp import dsl

@dsl.pipeline(
    name="Conditional Execution with Loop Pipeline",
    description="A pipeline that demonstrates conditional execution and looping."
)
def conditional_loop_pipeline(input_data_list: str, condition_list: str):
    input_data_list = json.loads(input_data_list)
    condition_list = json.loads(condition_list)
    with dsl.ParallelFor(input_data_list) as item:
        for condition in condition_list:
            process_data = process_data_conditional_component(item, condition)
if __name__ == "__main__":
    kfp.compiler.Compiler().compile(conditional_loop_pipeline, "conditional_loop_pipeline.yaml")

In this updated pipeline, we use the dsl.ParallelFor construct to loop over the input data list. For each item in the input data list, we loop over the condition list and call the process_data_conditional_component with the item and condition as arguments.

Step 5: Upload and Run the Pipeline with a List of Input Data and Conditions

Access the Kubeflow Pipelines dashboard by navigating to the URL provided during the setup process.
Click on the “Pipelines” tab in the left-hand sidebar.
Click the “Upload pipeline” button in the upper right corner.
In the “Upload pipeline” dialog, click “Browse” and select the conditional_loop_pipeline.yaml file generated in the previous step.
Click “Upload” to upload the pipeline to the Kubeflow platform.
Once the pipeline is uploaded, click on its name to open the pipeline details page.
Click the “Create run” button to start a new run of the pipeline.
On the “Create run” page, you can give your run a name and choose a pipeline version. Set the “input_data_list” and “condition_list” arguments to JSON-encoded lists of input data and conditions (e.g., ‘[“Hello, Kubeflow!”, “Machine Learning”]’ and ‘[“uppercase”, “lowercase”]’).
Click “Start” to begin the pipeline run.

In this tutorial, we covered how to implement conditional execution and loops in Kubeflow Pipelines using Python. With these advanced pipeline design techniques, you can create more complex and dynamic machine learning workflows, enabling greater flexibility and control over your ML experiments. As you continue to work with Kubeflow Pipelines, you can explore other advanced features to further enhance your machine learning workflows.

LyronFoster

Lyron Foster is a Hawai’i based African American Author, Musician, Actor, Blogger, Philanthropist and Multinational Serial Tech Entrepreneur.

lyronfoster.com

Containerizing Your Code: Docker and Kubeflow Pipelines

April 12, 2023 Machine Learning Workflows

Containerizing Your Code: Docker and Kubeflow Pipelines

Kubeflow Pipelines allows you to build, deploy, and manage end-to-end machine learning workflows. In order to use custom code in your pipeline, you need to containerize it using Docker. This ensures that your code can be easily deployed, scaled, and managed by Kubernetes, which is the underlying infrastructure for Kubeflow. In this tutorial, we will guide you through containerizing your Python code using Docker and integrating it into a Kubeflow Pipeline.

Prerequisites

Docker installed on your local machine (Download from https://www.docker.com/get-started)
Familiarity with Python programming
Kubeflow Pipelines installed and set up (follow our previous tutorial, “Setting up Kubeflow Pipelines: A Step-by-Step Guide”)

Step 1: Write Your Python Script

Create a new Python script (e.g., data_processing.py) containing the following code:

import sys

def process_data(input_data):
    return input_data.upper()
if __name__ == "__main__":
    input_data = sys.argv[1]
    processed_data = process_data(input_data)
    print(f"Processed data: {processed_data}")

This script takes an input string as a command-line argument, converts it to uppercase, and prints the result.

Step 2: Create a Dockerfile

Create a new file named Dockerfile in the same directory as your Python script, and add the following content:

FROM python:3.7

WORKDIR /app
COPY data_processing.py /app
ENTRYPOINT ["python", "data_processing.py"]

This Dockerfile specifies that the base image is python:3.7, sets the working directory to /app, copies the Python script into the container, and sets the entry point to execute the script when the container is run.

Step 3: Build the Docker Image

Open a terminal or command prompt, navigate to the directory containing the Dockerfile and Python script, and run the following command to build the Docker image:

docker build -t your_username/data_processing:latest .

Replace your_username with your Docker Hub username or another identifier. This command builds a Docker image with the specified tag and the current directory as the build context.

Step 4: Test the Docker Image

Test the Docker image by running the following command:

docker run --rm your_username/data_processing:latest "hello world"

This should output:

Processed data: HELLO WORLD

Step 5: Push the Docker Image to a Container Registry

To use the Docker image in a Kubeflow Pipeline, you need to push it to a container registry, such as Docker Hub, Google Container Registry, or Amazon Elastic Container Registry. In this tutorial, we will use Docker Hub.

First, log in to Docker Hub using the command line:

docker login

Enter your Docker Hub username and password when prompted.

Next, push the Docker image to Docker Hub:

docker push your_username/data_processing:latest

Step 6: Create a Kubeflow Pipeline using the Docker Image

Now that the Docker image is available in a container registry, you can use it in a Kubeflow Pipeline. Create a new Python script (e.g., custom_pipeline.py) and add the following code:

import kfp
from kfp import dsl

def data_processing_op(input_data: str):
    return dsl.ContainerOp(
        name="Data Processing",
        image="your_username/data_processing:latest",
        arguments=[input_data],
    )
@dsl.pipeline(
    name="Custom Pipeline",
    description="A pipeline that uses a custom Docker image for data processing."
)
def custom_pipeline(input_data: str = "hello world"):
    data_processing = data_processing_op(input_data)
if __name__ == "__main__":
    kfp.compiler.Compiler().compile(custom_pipeline, "custom_pipeline.yaml")

This Python script defines a pipeline with a single step that uses the custom Docker image we created earlier. The data_processing_op function takes an input string and returns a ContainerOp object with the specified Docker image and input data.

Step 7: Upload and Run the Pipeline

Access the Kubeflow Pipelines dashboard by navigating to the URL provided during the setup process.
Click on the “Pipelines” tab in the left-hand sidebar.
Click the “Upload pipeline” button in the upper right corner.
In the “Upload pipeline” dialog, click “Browse” and select the custom_pipeline.yaml file generated in the previous step.
Click “Upload” to upload the pipeline to the Kubeflow platform.
Once the pipeline is uploaded, click on its name to open the pipeline details page.
Click the “Create run” button to start a new run of the pipeline.
On the “Create run” page, you can give your run a name and choose a pipeline version. Click “Start” to begin the pipeline run.

Step 8: Monitor the Pipeline Run

After starting the pipeline run, you will be redirected to the “Run details” page. Here, you can monitor the progress of your pipeline, view the logs for each step, and inspect the output artifacts.

The pipeline graph will show the status of each step in the pipeline, with different colors indicating success, failure, or in-progress status.
To view the logs for a specific step, click on the step in the pipeline graph and then click the “Logs” tab in the right-hand pane.
To view the output artifacts, click on the step in the pipeline graph and then click the “Artifacts” tab in the right-hand pane.

Congratulations! You have successfully containerized your Python code using Docker and integrated it into a Kubeflow Pipeline. You can now leverage the power of containerization to build more complex pipelines with custom code, ensuring that your machine learning workflows are scalable, portable, and easily maintainable.

In this tutorial, we walked you through the process of containerizing your Python code using Docker and integrating it into a Kubeflow Pipeline. By using containers, you can ensure that your custom code is easily deployable, maintainable, and scalable across different environments. As you continue to work with Kubeflow Pipelines, you can explore more advanced features, build more sophisticated pipelines, and optimize your machine learning workflows.

LyronFoster

Lyron Foster is a Hawai’i based African American Author, Musician, Actor, Blogger, Philanthropist and Multinational Serial Tech Entrepreneur.

lyronfoster.com