This can be downloaded as part of the Wallaroo Tutorials repository.

Wallaroo ML Workload Orchestration Simple Tutorial

This tutorial demonstrates deploying multiple pipelines using Wallaroo ML Workload Orchestrations. For this use case, we look at a situation where multiple models are deployed, each which inference data from different regions.

Definitions

• Orchestration: A set of instructions written as a python script with a requirements library. Orchestrations are uploaded to the Wallaroo instance as a .zip file.
• Task: An implementation of an orchestration. Tasks are run either once when requested, on a repeating schedule, or as a service.
• Connection: Definitions set by MLOps engineers that are used by other Wallaroo users for connection information to a data source. Usually paired with orchestrations.

Tutorial Goals

The tutorial will demonstrate the following:

1. Create three pipelines with three different models as pipeline steps. Each model is trained to predict house price prices based on a series of inputs that include square footage, etc. These models are represented by the files:
   1. ./models/gbr_model.onnx
   2. ./models/rf_model.onnx
   3. ./models/xgb_model.onnx
2. The sample data is already converted into tensor format in the files:
   1. ./data/zonea.df.json
   2. ./data/zoneb.df.json
   3. ./data/zonec.df.json
3. Upload Wallaroo ML Workload Orchestration that:
   1. Deploys the three pipelines asynchronously.
   2. Test the pipeline status to verify that the status is Running.
   3. Once all three pipelines are Running, execute the sample inferences.
4. Create a Run Once task from the uploaded orchestration and show the Task Run logs to show the deployments and inferences executed successfully.

Prerequisites

• An installed Wallaroo instance.
• The following Python libraries installed. These are included by default in a Wallaroo instance’s JupyterHub service.
  • os
  • wallaroo: The Wallaroo SDK. Included with the Wallaroo JupyterHub service by default.
  • pandas: Pandas, mainly used for Pandas DataFrame.

Deploy and Serve

For this tutorial, we’ll create a workspace, upload our sample model and deploy a pipeline. We’ll perform some quick sample inferences to verify that everything it working.

Load Libraries

Here we’ll import the various libraries we’ll use for the tutorial.

import wallaroo
from wallaroo.object import EntityNotFoundError, RequiredAttributeMissing

# to display dataframe tables
from IPython.display import display
# used to display dataframe information without truncating
import pandas as pd
pd.set_option('display.max_colwidth', None)
pd.set_option('display.max_columns', None)

Connect to the Wallaroo Instance

The first step is to connect to Wallaroo through the Wallaroo client. The Python library is included in the Wallaroo install and available through the Jupyter Hub interface provided with your Wallaroo environment.

This is accomplished using the wallaroo.Client() command, which provides a URL to grant the SDK permission to your specific Wallaroo environment. When displayed, enter the URL into a browser and confirm permissions. Store the connection into a variable that can be referenced later.

If logging into the Wallaroo instance through the internal JupyterHub service, use wl = wallaroo.Client(). For more information on Wallaroo Client settings, see the Client Connection guide.

# Login through local Wallaroo instance

wl = wallaroo.Client()

# Setting variables for later steps

workspace_name = f'multiple_pipeline_deployment_tutorial'

model_name_a = 'house-price-model-a'
model_name_b = 'house-price-model-b'
model_name_c = 'house-price-model-c'

model_file_name_a = './models/gbr_model.onnx'
model_file_name_b = './models/rf_model.onnx'
model_file_name_c = './models/xgb_model.onnx'

pipeline_name_a = 'house-price-zone-a'
pipeline_name_b = 'house-price-zone-b'
pipeline_name_c = 'house-price-zone-c'

Create the Workspace

We’ll now create our workspace and pipeline for the tutorial. If this tutorial has been run previously, then this will retrieve the existing ones with the assumption they’re for us with this tutorial.

We’ll set the retrieved workspace as the current workspace in the SDK, so all commands will default to that workspace.

workspace = wl.get_workspace(name=workspace_name, create_if_not_exist=True)
wl.set_current_workspace(workspace)

{'name': 'multiple_pipeline_deployment_tutorial', 'id': 9, 'archived': False, 'created_by': 'c6fd6e38-db8c-4788-b6b1-dc882fd60e14', 'created_at': '2024-09-17T15:22:50.837949+00:00', 'models': [], 'pipelines': []}

Upload the Models

We’ll upload our model into our sample workspace, then add it as a pipeline step before deploying the pipeline to it’s ready to accept inference requests.

# Upload the models

housing_model_a = (wl.upload_model(model_name_a, 
                                    model_file_name_a, 
                                    framework=wallaroo.framework.Framework.ONNX)
                                    .configure(tensor_fields=["tensor"])
                        )

housing_model_b = (wl.upload_model(model_name_b, 
                                    model_file_name_b, 
                                    framework=wallaroo.framework.Framework.ONNX)
                                    .configure(tensor_fields=["tensor"])
                        )

housing_model_c = (wl.upload_model(model_name_c, 
                                    model_file_name_c, 
                                    framework=wallaroo.framework.Framework.ONNX)
                                    .configure(tensor_fields=["tensor"])
                        )

Create Pipelines and Add Models as Pipeline Steps

We will now create the pipelines, and add the the models as pipeline steps.

pipeline_a = wl.build_pipeline(pipeline_name_a)
pipeline_b = wl.build_pipeline(pipeline_name_b)
pipeline_c = wl.build_pipeline(pipeline_name_c)

pipeline_a.add_model_step(housing_model_a)
pipeline_b.add_model_step(housing_model_b)
pipeline_c.add_model_step(housing_model_c)

Simulate ML Orchestration Steps

The following code simulates how the ML Orchestration will behave:

• Retrieve the pipelines.
• Deploy the pipelines with the wait_for_status parameter, which skips waiting for the pipeline to complete deployment before moving to the next step.
• Verify the pipeline is deployed successfully before performing the inference requests.
• Submit separate inference requests for each deployed pipeline.
• Undeploy each pipeline.

# retrieve the pipelines

pipeline_a = wl.get_pipeline(pipeline_name_a)
pipeline_b = wl.get_pipeline(pipeline_name_b)
pipeline_c = wl.get_pipeline(pipeline_name_c)

# deploy the pipelines without waiting for the pipeline status

deploy_config = wallaroo.DeploymentConfigBuilder().replica_count(1).cpus(0.5).memory("1Gi").build()

pipeline_a.deploy(deployment_config=deploy_config, wait_for_status=False)
pipeline_b.deploy(deployment_config=deploy_config, wait_for_status=False)
pipeline_c.deploy(deployment_config=deploy_config, wait_for_status=False)

Deployment initiated for house-price-zone-a. Please check pipeline status.
Deployment initiated for house-price-zone-b. Please check pipeline status.
Deployment initiated for house-price-zone-c. Please check pipeline status.

# verify the pipeline is deployed

import time

while pipeline_a.status()['status'] != 'Running':
    time.sleep(15)
    print("Waiting for deployment.")

pipeline_a.status()

Waiting for deployment.
{‘status’: ‘Running’,

‘details’: [],

’engines’: [{‘ip’: ‘10.28.2.12’,

’name’: ’engine-5cf9547bf-bbnvb’,

‘status’: ‘Running’,

‘reason’: None,

‘details’: [],

‘pipeline_statuses’: {‘pipelines’: [{‘id’: ‘house-price-zone-a’,

‘status’: ‘Running’,

‘version’: ‘30e46049-6488-4ed5-b3fd-34764471e0b0’}]},

‘model_statuses’: {‘models’: [{‘model_version_id’: 1,

’name’: ‘house-price-model-a’,

‘sha’: ’ed6065a79d841f7e96307bb20d5ef22840f15da0b587efb51425c7ad60589d6a’,

‘status’: ‘Running’,

‘version’: ‘ade98e8e-1006-4931-8c7c-6b9e6efbc06b’}]}}],

’engine_lbs’: [{‘ip’: ‘10.28.2.11’,

’name’: ’engine-lb-6b59985857-57kl7’,

‘status’: ‘Running’,

‘reason’: None,

‘details’: []}],

‘sidekicks’: []}

# perform inference on the deployed pipelines

display(pipeline_a.infer_from_file('./data/zonea.df.json').head(20))
display(pipeline_b.infer_from_file('./data/zoneb.df.json').head(20))
display(pipeline_c.infer_from_file('./data/zonec.df.json').head(20))

## undeploy the pipeline

pipeline_a.undeploy()
pipeline_b.undeploy()
pipeline_c.undeploy()

Waiting for undeployment - this will take up to 45s .................................... ok
Waiting for undeployment - this will take up to 45s .................................... ok
Waiting for undeployment - this will take up to 45s .................................... ok

Multiple Pipeline ML Automation Example

With the pipeline deployed and our connections set, we will now generate our ML Workload Orchestration. See the Wallaroo ML Workload Orchestrations guide for full details.

Orchestrations are uploaded to the Wallaroo instance as a ZIP file with the following requirements:

For our example, our orchestration will:

1. Retrieve and deploy the pipelines.
2. Perform sample inferences stored in the .zip file as an artifact.

This sample script is stored in remote_inference/main.py with an empty requirements.txt file, and packaged into the orchestration as ./remote_inference/multiple_pieplines.zip. We’ll display the steps in uploading the orchestration to the Wallaroo instance.

Upload the Orchestration

Orchestrations are uploaded with the Wallaroo client upload_orchestration(path) method with the following parameters.

Once uploaded, the deployment will be prepared and any requirements will be downloaded and installed.

For this example, the orchestration ./remote_inference/remote_inference.zip will be uploaded and saved to the variable orchestration.

Orchestration Status

We will loop until the uploaded orchestration’s status displays ready.

orchestration = wl.upload_orchestration(path="./remote_inference/multiple_pipelines.zip")

while orchestration.status() != 'ready' and orchestration.status() != 'error':
    print(orchestration.status())
    time.sleep(5)

pending_packaging
pending_packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging

Create Task from Orchestration

Once an Orchestration has the status ready, it can be run as a task. Tasks have the following run options.

Run Task Once

We’ll do both a Run Once task and generate our Run Once Task from our orchestration.

Tasks are generated and run once with the Orchestration run_once(name, json_args, timeout) method. Any arguments for the orchestration are passed in as a Dict. If there are no arguments, then an empty set {} is passed.

# Example: run once

import datetime
task_start = datetime.datetime.now()
task = orchestration.run_once(name="multiple_pipelines_demo", 
                              json_args={})

Task Status

The list of tasks in the Wallaroo instance is retrieves through the Wallaroo Client list_tasks() method. This returns an array list of the following.

For this example, the status of the previously created task will be generated, then looped until it has reached status started.

while task.status() != "started":
    display(task.status())
    time.sleep(5)
task.status()

'pending'
‘pending’
‘started’

Task Results

We can view the inferences from our logs and verify that new entries were added from our task. We can do that with the task last_runs() method to see the list of task runs executed, then show the log from the last completed task run.

# wait for the deployment and inference to complete
time.sleep(45)
task.last_runs()[0]

task.last_runs()[0].logs()

2024-17-Sep 18:04:37 Getting the workspace multiple_pipeline_deployment_tutorial
2024-17-Sep 18:04:37 Getting the pipelines
2024-17-Sep 18:04:37 Deployment initiated for house-price-zone-a. Please check pipeline status.
2024-17-Sep 18:04:37 Deployment initiated for house-price-zone-b. Please check pipeline status.
2024-17-Sep 18:04:37 Deployment initiated for house-price-zone-c. Please check pipeline status.
2024-17-Sep 18:04:37                       time  ... anomaly.count
2024-17-Sep 18:04:37 0  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 1  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 2  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 4  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 3  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 5  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 7  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 6  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 8  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 9  2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 10 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 11 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 12 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 13 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 14 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 16 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 15 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 17 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 19 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 18 2024-09-17 18:02:45.072  ...             0
2024-17-Sep 18:04:37 
2024-17-Sep 18:04:37                       time  ... anomaly.count
2024-17-Sep 18:04:37 [20 rows x 4 columns]
2024-17-Sep 18:04:37 0  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 2  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 1  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 3  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 4  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 6  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 5  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 7  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 8  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 10 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 9  2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 11 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 13 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 12 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 14 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 16 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 15 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 17 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 19 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 18 2024-09-17 18:02:45.129  ...             0
2024-17-Sep 18:04:37 [20 rows x 4 columns]
2024-17-Sep 18:04:37 
2024-17-Sep 18:04:37                       time  ... anomaly.count
2024-17-Sep 18:04:37 0  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 1  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 2  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 3  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 4  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 5  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 6  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 7  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 9  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 10 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 8  2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 11 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 12 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 13 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 14 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 15 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 16 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 17 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 18 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 19 2024-09-17 18:02:45.216  ...             0
2024-17-Sep 18:04:37 
2024-17-Sep 18:04:37 [20 rows x 4 columns]

This completes the tutorial and example. The sample code and orchestration is available for organizations to use for deploying multiple pipelines and other tasks.