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

ML Workload Orchestration Comprehensive Tutorial

This tutorial provides a complete set of methods and examples regarding Wallaroo Connections and Wallaroo ML Workload Orchestration.

Wallaroo provides data connections, orchestrations, and tasks to provide organizations with a method of creating and managing automated tasks that can either be run on demand, on a regular schedule, or as a service so they respond to requests.

A typical flow in the orchestration, task and connection life cycle is:

1. (Optional) A connection is defined with information such as username, connection URL, tokens, etc.
2. One or more connections are applied to a workspace for users to implement in their code or orchestrations.
3. An orchestration is created to perform some set instructions. For example:
   1. Deploy a pipeline, request data from an external service, store the results in an external database, then undeploy the pipeline.
   2. Download a ML Model then replace a current pipeline step with the new version.
   3. Collect log files from a deployed pipeline once every hour and submit it to a Kafka or other service.
4. A task is created that specifies the orchestration to perform and the schedule:
   1. Run once.
   2. Run on a schedule (based on cron like settings).
   3. Run as a service to be run whenever requested.
5. Once the use for a task is complete, it is killed and its schedule or service removed.

Tutorial Goals

The tutorial will demonstrate the following:

1. Create a simple connection to retrieve an Apache Arrow table file from a GitHub registry.
2. Create an orchestration that retrieves the Apache Arrow table file from the location defined by the connection, deploy a pipeline, perform an inference, then undeploys the pipeline.
3. Implement the orchestration as a task that runs every minute.
4. Display the logs from the pipeline after 5 minutes to verify the task is running.

Tutorial Required Libraries

The following libraries are required for this tutorial, and included by default in a Wallaroo instance’s JupyterHub service.

• IMPORTANT NOTE: These libraries are already installed in the Wallaroo JupyterHub service. Do not uninstall and reinstall the Wallaroo SDK with the command below.

• wallaroo: The Wallaroo SDK.

• pandas: The pandas data analysis library.

• pyarrow: The Apache Arrow Python library.

The specific versions used are set in the file ./resources/requirements.txt. Supported libraries are automatically installed with the pypi or conda commands. For example, from the root of this tutorials folder:

pip install -r ./resources/requirements.txt

Initialization

The first step is to connect to a Wallaroo instance. We’ll load the libraries and set our client connection settings

Workspace, Model and Pipeline Setup

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.

import wallaroo
from wallaroo.object import EntityNotFoundError

# 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)
import pyarrow as pa

import requests

# Used to create unique workspace and pipeline names
import string
import random

# make a random 4 character suffix
suffix= ''.join(random.choice(string.ascii_lowercase) for i in range(4))
display(suffix)

'tgiq'

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'orchestrationworkspace{suffix}'
pipeline_name = f'orchestrationpipeline{suffix}'
model_name = f'orchestrationmodel{suffix}'
model_file_name = './models/rf_model.onnx'
connection_name = f'houseprice_arrow_table{suffix}'

Helper Methods

The following helper methods are used to either create or get workspaces and pipelines.

# helper methods to retrieve workspaces and pipelines

def get_workspace(name):
    workspace = None
    for ws in wl.list_workspaces():
        if ws.name() == name:
            workspace= ws
    if(workspace == None):
        workspace = wl.create_workspace(name)
    return workspace

def get_pipeline(name):
    try:
        pipeline = wl.pipelines_by_name(name)[0]
    except EntityNotFoundError:
        pipeline = wl.build_pipeline(name)
    return pipeline

Create the Workspace and Pipeline

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 = get_workspace(workspace_name)
wl.set_current_workspace(workspace)

pipeline = get_pipeline(pipeline_name)

Upload the Model and Deploy Pipeline

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 model

housing_model_control = wl.upload_model(model_name, model_file_name, framework=wallaroo.framework.Framework.ONNX).configure()

# Add the model as a pipeline step

pipeline.add_model_step(housing_model_control)

#deploy the pipeline
pipeline.deploy()

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

Sample Inferences

We’ll perform some quick sample inferences using an Apache Arrow table as the input. Once that’s finished, we’ll undeploy the pipeline and return the resources back to the Wallaroo instance.

# sample inferences

batch_inferences = pipeline.infer_from_file('./data/xtest-1k.arrow')

large_inference_result =  batch_inferences.to_pandas()
display(large_inference_result.head(20))

Create Wallaroo Connection

Connections are created at the Wallaroo instance level, typically by a MLOps or DevOps engineer, then applied to a workspace.

For this section:

1. We will create a sample connection that just has a URL to the same Arrow table file we used in the previous step.
2. We’ll apply the data connection to the workspace above.
3. For a quick demonstration, we’ll use the connection to retrieve the Arrow table file and use it for a quick sample inference.

Create Connection

Connections are created with the Wallaroo client command create_connection with the following parameters.

We’ll create the connection named houseprice_arrow_table, set it to the type HTTPFILE, and provide the details as 'host':'https://github.com/WallarooLabs/Wallaroo_Tutorials/raw/main/wallaroo-testing-tutorials/houseprice-saga/data/xtest-1k.arrow' - the location for our sample Arrow table inference input.

wl.create_connection(connection_name, 
                  "HTTPFILE", 
                  {'host':'https://github.com/WallarooLabs/Wallaroo_Tutorials/raw/main/wallaroo-testing-tutorials/houseprice-saga/data/xtest-1k.arrow'}
                  )

List Data Connections

The Wallaroo Client list_connections() method lists all connections for the Wallaroo instance.

wl.list_connections()

Add Connection to Workspace

The method Workspace add_connection(connection_name) adds a Data Connection to a workspace, and takes the following parameters.

We’ll add this connection to our sample workspace.

workspace.add_connection(connection_name)

Get Connection

Connections are retrieved by the Wallaroo Client get_connection(name) method.

connection = wl.get_connection(connection_name)

Connection Details

The Connection method details() retrieves a the connection details() as a dict.

display(connection.details())

{'host': 'https://github.com/WallarooLabs/Wallaroo_Tutorials/raw/main/wallaroo-testing-tutorials/houseprice-saga/data/xtest-1k.arrow'}

Using a Connection Example

For this example, the connection will be used to retrieve the Apache Arrow file referenced in the connection, and use that to turn it into an Apache Arrow table, then use that for a sample inference.

# Deploy the pipeline 
pipeline.deploy()

# Retrieve the file
# set accept as apache arrow table
headers = {
    'Accept': 'application/vnd.apache.arrow.file'
}

response = requests.get(
                    connection.details()['host'], 
                    headers=headers
                )

# Arrow table is retrieved 
with pa.ipc.open_file(response.content) as reader:
    arrow_table = reader.read_all()

results = pipeline.infer(arrow_table)

result_table = results.to_pandas()
display(result_table.head(20))

 ok

Remove Connection from Workspace

The Workspace method remove_connection(connection_name) removes the connection from the workspace, but does not delete the connection from the Wallaroo instance. This method takes the following parameters.

The previous connection will be removed from the workspace, then the workspace connections displayed to verify it has been removed.

workspace.remove_connection(connection_name)

display(workspace.list_connections())

(no connections)

Delete Connection

The Connection method delete_connection() removes the connection from the Wallaroo instance, and all attachments in workspaces they were connected to.

connection.delete_connection()

wl.list_connections()

(no connections)

Orchestration Tutorial

The next series of examples will build on what we just did. So far we have:

• Deployed a pipeline, performed sample inferences with a local Apache Arrow file, displayed the results, then undeployed the pipeline.
• Deployed a pipeline, use a Wallaroo connection details to retrieve a remote Apache Arrow file, performed inferences and displayed the results, then undeployed the pipeline.

For the orchestration tutorial, we’ll do the same thing only package it into a separate python script and upload it to the Wallaroo instance, then create a task from that orchestration and perform our sample inferences again.

Orchestration Requirements

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

• The ZIP file should not contain any directories - only files at the top level.

Zip Instructions

In a terminal with the zip command, assemble artifacts as above and then create the archive. The zip command is included by default with the Wallaroo JupyterHub service.

zip commands take the following format, with {zipfilename}.zip as the zip file to save the artifacts to, and each file thereafter as the files to add to the archive.

zip {zipfilename}.zip file1, file2, file3....

For example, the following command will add the files main.py and requirements.txt into the file hello.zip.

$ zip hello.zip main.py requirements.txt 
  adding: main.py (deflated 47%)
  adding: requirements.txt (deflated 52%)

Orchestration Recommendations

The following recommendations will make using Wallaroo orchestrations

• The version of Python used should match the same version as in the Wallaroo JupyterHub service.
• The same version of the Wallaroo SDK should match the server. For a 2023.2 Wallaroo instance, use the Wallaroo SDK version 2023.2.
• Specify the version of pip dependencies.
• The wallaroo.Client constructor auth_type argument is ignored. Using wallaroo.Client() is sufficient.
• The following methods will assist with orchestrations:
  • wallaroo.in_task() : Returns True if the code is running within an Orchestrator task.
  • wallaroo.task_args(): Returns a Dict of invocation-specific arguments passed to the run_ calls.
• Use print commands so outputs are saved to the task’s log files.

Example requirements.txt file

dbt-bigquery==1.4.3
dbt-core==1.4.5
dbt-extractor==0.4.1
dbt-postgres==1.4.5
google-api-core==2.8.2
google-auth==2.11.0
google-auth-oauthlib==0.4.6
google-cloud-bigquery==3.3.2
google-cloud-bigquery-storage==2.15.0
google-cloud-core==2.3.2
google-cloud-storage==2.5.0
google-crc32c==1.5.0
google-pasta==0.2.0
google-resumable-media==2.3.3
googleapis-common-protos==1.56.4

Sample Orchestrator

The following orchestrator artifacts are in the directory ./remote_inference and includes the file main.py with the following code:

import wallaroo
from wallaroo.object import EntityNotFoundError
import pandas as pd
import pyarrow as pa
import requests

wl = wallaroo.Client()

# Setting variables for later steps

# get the arguments
arguments = wl.task_args()

if "workspace_name" in arguments:
    workspace_name = arguments['workspace_name']
else:
    workspace_name="orchestrationworkspace"

if "pipeline_name" in arguments:
    pipeline_name = arguments['pipeline_name']
else:
    pipeline_name="orchestrationpipeline"

if "connection_name" in arguments:
    connection_name = arguments['connection_name']
else:
    connection_name = "houseprice_arrow_table"

# helper methods to retrieve workspaces and pipelines

def get_workspace(name):
    workspace = None
    for ws in wl.list_workspaces():
        if ws.name() == name:
            workspace= ws
    if(workspace == None):
        workspace = wl.create_workspace(name)
    return workspace

def get_pipeline(name):
    try:
        pipeline = wl.pipelines_by_name(name)[0]
    except EntityNotFoundError:
        pipeline = wl.build_pipeline(name)
    return pipeline

print(f"Getting the workspace {workspace_name}")
workspace = get_workspace(workspace_name)
wl.set_current_workspace(workspace)

print(f"Getting the pipeline {pipeline_name}")
pipeline = get_pipeline(pipeline_name)
pipeline.deploy()
# Get the connection - assuming it will be the only one

inference_source_connection = wl.get_connection(name=connection_name)

print(f"Getting arrow table file")
# Retrieve the file
# set accept as apache arrow table
headers = {
    'Accept': 'application/vnd.apache.arrow.file'
}

response = requests.get(
                    inference_source_connection.details()['host'], 
                    headers=headers
                )

# Arrow table is retrieved 
with pa.ipc.open_file(response.content) as reader:
    arrow_table = reader.read_all()

print("Inference time.  Displaying results after.")
# Perform the inference
result = pipeline.infer(arrow_table)
print(result)

pipeline.undeploy()

This is saved to the file ./remote_inference/remote_inference.zip.

Preparing the Wallaroo Instance

To prepare the Wallaroo instance, we’ll once again create the Wallaroo connection houseprice_arrow_table and apply it to the workspace.

wl.create_connection(connection_name, 
                  "HTTPFILE", 
                  {'host':'https://github.com/WallarooLabs/Wallaroo_Tutorials/raw/main/wallaroo-testing-tutorials/houseprice-saga/data/xtest-1k.arrow'}
                  )

workspace.add_connection(connection_name)

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 = wl.upload_orchestration(name="comprehensive sample", path="./remote_inference/remote_inference.zip")

Orchestration Status

The Orchestration method status() displays the current status of the uploaded orchestration.

For this example, the status of the orchestration will be displayed then looped until it has reached status ready.

import time

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

pending_packaging
pending_packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging

List Orchestrations

Orchestrations are listed with the Wallaroo Client list_orchestrations() which returns a list of available orchestrations.

wl.list_orchestrations()

Task Management Tutorial

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

Task Run Once

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.

For our example, we will pass the workspace, pipeline, and connection into our task.

# Example: run once

import datetime
task_start = datetime.datetime.now()

task = orchestration.run_once(name="house price run once 2", json_args={"workspace_name": workspace_name, 
                                                                           "pipeline_name":pipeline_name,
                                                                           "connection_name": connection_name
                                                                           }
                            )
task

taskfail.last_runs()

List Tasks

The list of tasks in the Wallaroo instance is retrieved through the Wallaroo Client list_tasks() method that accepts the following parameters.

This returns an array list of the following in reverse chronological order from updated at.

wl.list_tasks()

Task Status

The status of the task is returned with the Task status() method that returned the tasks status. Tasks can have the following status.

• pending: The task has not been started or is being prepared.
• started: The task has started to execute.

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

Task Last Runs History

The history of a task, which each deployment of the task is known as a task run is retrieved with the Task last_runs method that takes the following arguments.

This returns the following in reverse chronological order by updated at.

task.last_runs()

Task Run Logs

The output of a task is displayed with the Task Run logs() method that takes the following parameters.

The Task Run logs() returns the log entries as a string list, with each entry as an item in the list.

• IMPORTANT NOTE: It may take around a minute for task run logs to be integrated into the Wallaroo log database.

# give time for the task to complete and the log files entered
time.sleep(60)
recent_run = task.last_runs()[0]
display(recent_run.logs())

2023-22-May 19:59:29 Getting the workspace orchestrationworkspacetgiq
2023-22-May 19:59:29 Getting the pipeline orchestrationpipelinetgiq
2023-22-May 19:59:29 Getting arrow table file
2023-22-May 19:59:29 Inference time.  Displaying results after.
2023-22-May 19:59:29 pyarrow.Table
2023-22-May 19:59:29 time: timestamp[ms]
2023-22-May 19:59:29 in.tensor: list not null
2023-22-May 19:59:29   child 0, item: float
2023-22-May 19:59:29 out.variable: list not null
2023-22-May 19:59:29 check_failures: int8
2023-22-May 19:59:29   child 0, inner: float not null
2023-22-May 19:59:29 ----
2023-22-May 19:59:29 time: [[2023-05-22 19:58:49.767,2023-05-22 19:58:49.767,2023-05-22 19:58:49.767,2023-05-22 19:58:49.767,2023-05-22 19:58:49.767,...,2023-05-22 19:58:49.767,2023-05-22 19:58:49.767,2023-05-22 19:58:49.767,2023-05-22 19:58:49.767,2023-05-22 19:58:49.767]]
2023-22-May 19:59:29 in.tensor: [[[4,2.5,2900,5505,2,...,2970,5251,12,0,0],[2,2.5,2170,6361,1,...,2310,7419,6,0,0],...,[3,1.75,2910,37461,1,...,2520,18295,47,0,0],[3,2,2005,7000,1,...,1750,4500,34,0,0]]]
2023-22-May 19:59:29 check_failures: [[0,0,0,0,0,...,0,0,0,0,0]]
2023-22-May 19:59:29 out.variable: [[[718013.75],[615094.56],...,[706823.56],[581003]]]

Failed Task Logs

We can create a task that fails and show it in the last_runs list, then retrieve the logs to display why it failed.

# Example: run once

import datetime
task_start = datetime.datetime.now()

taskfail = orchestration.run_once(name="house price run once 2", json_args={"workspace_name": "bob", 
                                                                           "pipeline_name":"does not exist",
                                                                           "connection_name": connection_name
                                                                           }
                            )

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

# time.sleep(60)
taskfail.last_runs()

# time.sleep(60)
taskfaillogs = taskfail.last_runs()[0].logs()
display(taskfaillogs)

2023-22-May 20:17:22 Getting the workspace bob
2023-22-May 20:17:22   File "/home/jovyan/main.py", line 43, in get_pipeline
2023-22-May 20:17:22 Traceback (most recent call last):
2023-22-May 20:17:22 Getting the pipeline does not exist
2023-22-May 20:17:22     pipeline = wl.pipelines_by_name(name)[0]
2023-22-May 20:17:22   File "/home/jovyan/venv/lib/python3.9/site-packages/wallaroo/client.py", line 1064, in pipelines_by_name
2023-22-May 20:17:22     raise EntityNotFoundError("Pipeline", {"pipeline_name": pipeline_name})
2023-22-May 20:17:22 wallaroo.object.EntityNotFoundError: Pipeline not found: {'pipeline_name': 'does not exist'}
2023-22-May 20:17:22 
2023-22-May 20:17:22 During handling of the above exception, another exception occurred:
2023-22-May 20:17:22 Traceback (most recent call last):
2023-22-May 20:17:22 
2023-22-May 20:17:22   File "/home/jovyan/main.py", line 54, in 
2023-22-May 20:17:22     pipeline = get_pipeline(pipeline_name)
2023-22-May 20:17:22     pipeline = wl.build_pipeline(name)
2023-22-May 20:17:22   File "/home/jovyan/main.py", line 45, in get_pipeline
2023-22-May 20:17:22   File "/home/jovyan/venv/lib/python3.9/site-packages/wallaroo/client.py", line 1102, in build_pipeline
2023-22-May 20:17:22     require_dns_compliance(pipeline_name)
2023-22-May 20:17:22   File "/home/jovyan/venv/lib/python3.9/site-packages/wallaroo/checks.py", line 274, in require_dns_compliance
2023-22-May 20:17:22 wallaroo.object.InvalidNameError: Name 'does not exist is invalid: must be DNS-compatible (ASCII alpha-numeric plus dash (-))
2023-22-May 20:17:22     raise InvalidNameError(

Task Results

We can view the inferences from our logs and verify that new entries were added from our task. In our case, we’ll assume the task once started takes about 1 minute to run (deploy the pipeline, run the inference, undeploy the pipeline). We’ll add in a wait of 1 minute, then display the logs during the time period the task was running.

task_end = datetime.datetime.now()
display(task_end)

pipeline.logs(start_datetime = task_start, end_datetime = task_end)

datetime.datetime(2023, 5, 22, 20, 1, 25, 418564)
Warning: Pipeline log size limit exceeded. Please request logs using export_logs

506 rows × 4 columns

Scheduled Tasks

Scheduled tasks are run with the Orchestration run_scheduled method. We’ll set it up to run every 5 minutes, then check the results.

It is recommended that orchestrations that have pipeline deploy or undeploy commands be spaced out no less than 5 minutes to prevent colliding with other tasks that use the same pipeline.

task_start = datetime.datetime.now()
schedule = "*/5 * * * *"
task_scheduled = orchestration.run_scheduled(name="schedule example", 
                                             timeout=600, 
                                             schedule=schedule, 
                                             json_args={"workspace_name": workspace_name, 
                                                        "pipeline_name": pipeline_name,
                                                        "connection_name": connection_name
                                            })
while task_scheduled.status() != "started":
    display(task_scheduled.status())
    time.sleep(5)
task_scheduled

'pending'

time.sleep(420)
recent_run = task_scheduled.last_runs()[0]
display(recent_run.logs())

2023-22-May 20:11:02 Getting the workspace orchestrationworkspacetgiq
2023-22-May 20:11:02 Getting the pipeline orchestrationpipelinetgiq
2023-22-May 20:11:02 Inference time.  Displaying results after.
2023-22-May 20:11:02 Getting arrow table file
2023-22-May 20:11:02 pyarrow.Table
2023-22-May 20:11:02 time: timestamp[ms]
2023-22-May 20:11:02 in.tensor: list not null
2023-22-May 20:11:02   child 0, item: float
2023-22-May 20:11:02 out.variable: list not null
2023-22-May 20:11:02   child 0, inner: float not null
2023-22-May 20:11:02 check_failures: int8
2023-22-May 20:11:02 ----
2023-22-May 20:11:02 time: [[2023-05-22 20:10:23.271,2023-05-22 20:10:23.271,2023-05-22 20:10:23.271,2023-05-22 20:10:23.271,2023-05-22 20:10:23.271,...,2023-05-22 20:10:23.271,2023-05-22 20:10:23.271,2023-05-22 20:10:23.271,2023-05-22 20:10:23.271,2023-05-22 20:10:23.271]]
2023-22-May 20:11:02 in.tensor: [[[4,2.5,2900,5505,2,...,2970,5251,12,0,0],[2,2.5,2170,6361,1,...,2310,7419,6,0,0],...,[3,1.75,2910,37461,1,...,2520,18295,47,0,0],[3,2,2005,7000,1,...,1750,4500,34,0,0]]]
2023-22-May 20:11:02 check_failures: [[0,0,0,0,0,...,0,0,0,0,0]]
2023-22-May 20:11:02 out.variable: [[[718013.75],[615094.56],...,[706823.56],[581003]]]

Kill a Task

Killing a task removes the schedule or removes it from a service. Tasks are killed with the Task kill() method, and returns a message with the status of the kill procedure.

If necessary, all tasks can be killed through the following script.

• IMPORTANT NOTE: This command will kill all running tasks - scheduled or otherwise. Only use this if required.

# Kill all tasks
for t in wl.list_tasks(): t.kill()

When listed with Wallaroo client task_list(killed=True) , the field active displays tasks that are killed (False) or either completed running or still scheduled to run (True).

task_scheduled.kill()

<ArbexStatus.PENDING_KILL: 'pending_kill'>

wl.list_tasks(killed=True)

Cleaning Up

With the tutorial complete we will undeploy the pipeline and ensure the resources are returned back to the Wallaroo instance.

pipeline.undeploy()