ML Workload Orchestration Tutorials

• 1: Wallaroo ML Workload Orchestration API Tutorial
• 2: Wallaroo Connection API with Google BigQuery Tutorial
• 3: Wallaroo ML Workload Orchestration Simple Tutorial
• 4: Wallaroo ML Workload Orchestration Google BigQuery with House Price Model Tutorial
• 5: Wallaroo ML Workload Orchestration Google BigQuery with Statsmodel Forecast Tutorial
• 6: Wallaroo ML Workload Orchestration Comprehensive Tutorial

1 - Wallaroo ML Workload Orchestration API Tutorial

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

Wallaroo Connection and ML Workload Orchestration API Tutorial

This tutorial provides a quick set of methods and examples regarding Wallaroo Connections and Wallaroo ML Workload Orchestration. For full details, see the Wallaroo Documentation site.

Wallaroo provides Data Connections and ML Workload Orchestrations to provide organizations with a method of creating and managing automated tasks that can either be run on demand or a regular schedule.

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 a workspace and pipeline with a sample model.
2. Upload Wallaroo ML Workload Orchestration through the Wallaroo MLOps API.
3. List available orchestrations through the Wallaroo MLOps API.
4. Run the orchestration once as a Run Once Task through the Wallaroo MLOps API and verify that the information was saved the pipeline logs.

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
  • pyarrow: PyArrow for Apache Arrow support

Initial Steps

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

import time

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)

'gsze'

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(). If logging in externally, update the wallarooPrefix and wallarooSuffix variables with the proper DNS information. For more information on Wallaroo Client settings, see the Client Connection guide.

# Login through local Wallaroo instance

wl = wallaroo.Client()

API URL

The variable APIURL is used to specify the connection to the Wallaroo instance’s MLOps API URL, and is composed of the Wallaroo DNS prefix and suffix. For full details, see the Wallaroo API Connection Guide
.

The variables wallarooPrefix and wallarooSuffix variables will be used to derive the API url. For example, if the Wallaroo Prefix is doc-test and the url is example.com, then the MLOps API URL would be doc-test.api.example.com/v1/api/{request}.

Note the . is part of the prefix. If there is no prefix, then wallarooPrefix = ""

Set the Wallaroo Prefix and Suffix in the code segment below based on your Wallaroo instance.

# Setting variables for later steps

wallarooPrefix = "YOUR PREFIX."

wallarooSuffix = "YOUR SUFFIX"

APIURL = f"https://{wallarooPrefix}api.{wallarooSuffix}"

workspace_name = f'apiorchestrationworkspace{suffix}'
pipeline_name = f'apipipeline{suffix}'
model_name = f'apiorchestrationmodel{suffix}'
model_file_name = './models/rf_model.onnx'

Helper Methods

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

# 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

def get_connection(name, connection_type, connection_arguments):
    try:
        connection = wl.get_connection(name)
    except RequiredAttributeMissing:
        connection =wl.create_connection(name, 
                  connection_type, 
                  connection_arguments)
    return connection

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)

workspace_id = workspace.id()

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(tensor_fields=["tensor"])
                        )

# 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

Wallaroo ML Workload Orchestration 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. Use the inference_results_connection to open a HTTP Get connection to the inference data file and use it in an inference request in the deployed pipeline.
2. Submit the inference results to the location specified in the external_inference_connection.

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

Note that the orchestration assumes the pipeline is already deployed.

API Upload the Orchestration

Orchestrations are uploaded via POST as a application/octet-stream with MLOps API route:

• REQUEST
  • POST /v1/api/orchestration/upload
  • Content-Type multipart/form-data
• PARAMETERS
  • file: The file uploaded as Content-Type as application/octet-stream.
  • metadata: Included as Content-Type as application/json with:
    • workspace_id: The numerical id of the workspace to upload the orchestration to.
    • name: The name of the orchestration.
    • The metadata specifying the workspace id and Content-Type as application/json.

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

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/orchestration/upload"

fp = open("./api_inference_orchestration.zip", "rb")

metadata = f'{{"workspace_id": {workspace_id},"name": "apiorchestrationsample"}}'

response = requests.post(
    url,
    headers=headers,
    files=[
        ("file", 
            ("api_inference_orchestration.zip", fp, "application/octet-stream")
        ),
        ("metadata", 
            ("metadata", metadata, "application/json")
        )
    ],
).json()

display(response)
orchestration_id = response['id']

{'id': 'b951f7b8-0690-4004-86bf-cc9802359313'}

API List Orchestrations

A list of orchestrations retrieved via POST MLOps API route:

• REQUEST
  • POST /v1/api/orchestration/list
• PARAMETERS
  • workspace_id: The numerical identifier of the workspace associated with the orchestration.

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/orchestration/list"

data = {
    'workspace_id': workspace_id
}

response=requests.post(url, headers=headers, json=data)
display(response.json())

[{'id': 'b951f7b8-0690-4004-86bf-cc9802359313',
  'workspace_id': 8,
  'sha': 'd3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e',
  'name': 'apiorchestrationsample',
  'file_name': 'api_inference_orchestration.zip',
  'task_id': 'c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef',
  'owner_id': 'a6857628-b0aa-451e-a0a0-bbc1d6eea6e0',
  'status': 'pending_packaging',
  'created_at': '2023-05-22T20:48:41.233482+00:00',
  'updated_at': '2023-05-22T20:48:41.233482+00:00'}]

API Get Orchestration

A list of orchestrations retrieved via POST MLOps API route:

• REQUEST
  • POST /v1/api/orchestration/get_by_id
• PARAMETERS
  • id: The UUID of the orchestration being retrieved.
• RETURNS
  • id: The ID of the orchestration in UUID format.
  • workspace_id: Numerical value of the workspace the orchestration was uploaded to.
  • sha: The SHA hash value of the orchestration.
  • file_name: The file name the orchestration was uploaded as.
  • task_id: The task id in UUID format for unpacking and preparing the orchestration.
  • owner_id: The Keycloak ID of the user that uploaded the orchestration.
  • status: The status of the orchestration. Status values are:
    • packing: Preparing the orchestration to be used as a task.
    • ready: The orchestration is ready to be deployed as a task.

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/orchestration/get_by_id"

data = {
    'id': orchestration_id
}

# loop until status is ready
status = None

while status != 'ready':
    response=requests.post(url, headers=headers, json=data).json()
    display(response)
    status = response['status']
    time.sleep(10)

orchestration_sha = response['sha']

{'id': 'b951f7b8-0690-4004-86bf-cc9802359313',
 'workspace_id': 8,
 'sha': 'd3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e',
 'file_name': 'api_inference_orchestration.zip',
 'task_id': 'c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef',
 'owner_id': 'a6857628-b0aa-451e-a0a0-bbc1d6eea6e0',
 'status': 'packaging'}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘packaging’}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘packaging’}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘packaging’}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘packaging’}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘packaging’}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘packaging’}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘packaging’}
{‘id’: ‘b951f7b8-0690-4004-86bf-cc9802359313’,

‘workspace_id’: 8,

‘sha’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘file_name’: ‘api_inference_orchestration.zip’,

’task_id’: ‘c9622cf8-cbe5-4e3c-b64c-dabd6c5b7fef’,

‘owner_id’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘status’: ‘ready’}

Task Management Tutorial

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

Run Task Once via API

We’ll do both a Run Once task and generate our Run Once Task from our orchestration. Orchestrations are started as a run once task with the following request:

• REQUEST
  • POST /v1/api/orchestration/task/run_once
• PARAMETERS
  • name (String Required): The name to assign to the task.
  • workspace_id (Integer Required): The numerical identifier of the workspace associated with the orchestration.
  • orch_id(String Required): The orchestration ID represented by a UUID.
  • json(Dict Required): The parameters to pass to the task.

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

# retrieve the authorization token
headers = wl.auth.auth_header()

data = {
    "name": "api run once task",
    "workspace_id": workspace_id,
    "orch_id": orchestration_id,
    "json": {
        "workspace_name": workspace_name,
        "pipeline_name": pipeline_name
    }
}

import datetime
task_start = datetime.datetime.now()

url=f"{APIURL}/v1/api/task/run_once"

response=requests.post(url, headers=headers, json=data).json()
display(response)
task_id = response['id']

{'id': 'c868aa44-f7fe-4e3d-b11d-e1e6af3ec150'}

Task Status via API

The list of tasks in the Wallaroo instance is retrieves through the Wallaroo MLOPs API request:

• REQUEST
  • POST /v1/api/task/get_by_id
• PARAMETERS
  • task: The numerical identifier of the workspace associated with the orchestration.
  • orch_id: The orchestration ID represented by a UUID.
  • json: The parameters to pass to the task.

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

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/get_by_id"

data = {
    "id": task_id
}

status = None

while status != 'started':
    response=requests.post(url, headers=headers, json=data).json()
    display(response)
    status = response['status']
    time.sleep(10)

{'name': 'api run once task',
 'id': 'c868aa44-f7fe-4e3d-b11d-e1e6af3ec150',
 'image': 'proxy.replicated.com/proxy/wallaroo/ghcr.io/wallaroolabs/arbex-orch-deploy',
 'image_tag': 'v2023.2.0-main-3271',
 'bind_secrets': ['minio'],
 'extra_env_vars': {'MINIO_URL': 'http://minio.wallaroo.svc.cluster.local:9000',
  'ORCH_OWNER_ID': 'a6857628-b0aa-451e-a0a0-bbc1d6eea6e0',
  'ORCH_SHA': 'd3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e',
  'TASK_DEBUG': 'false',
  'TASK_ID': 'c868aa44-f7fe-4e3d-b11d-e1e6af3ec150'},
 'auth_init': True,
 'workspace_id': 8,
 'flavor': 'exec_orch_oneshot',
 'reap_threshold_secs': 900,
 'exec_type': 'job',
 'status': 'pending',
 'input_data': {'pipeline_name': 'apipipelinegsze',
  'workspace_name': 'apiorchestrationworkspacegsze'},
 'killed': False,
 'created_at': '2023-05-22T21:08:31.099447+00:00',
 'updated_at': '2023-05-22T21:08:31.105312+00:00',
 'last_runs': []}
{’name’: ‘api run once task’,

‘id’: ‘c868aa44-f7fe-4e3d-b11d-e1e6af3ec150’,

‘image’: ‘proxy.replicated.com/proxy/wallaroo/ghcr.io/wallaroolabs/arbex-orch-deploy’,

‘image_tag’: ‘v2023.2.0-main-3271’,

‘bind_secrets’: [‘minio’],

’extra_env_vars’: {‘MINIO_URL’: ‘http://minio.wallaroo.svc.cluster.local:9000’,

‘ORCH_OWNER_ID’: ‘a6857628-b0aa-451e-a0a0-bbc1d6eea6e0’,

‘ORCH_SHA’: ‘d3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e’,

‘TASK_DEBUG’: ‘false’,

‘TASK_ID’: ‘c868aa44-f7fe-4e3d-b11d-e1e6af3ec150’},

‘auth_init’: True,

‘workspace_id’: 8,

‘flavor’: ’exec_orch_oneshot’,

‘reap_threshold_secs’: 900,

’exec_type’: ‘job’,

‘status’: ‘started’,

‘input_data’: {‘pipeline_name’: ‘apipipelinegsze’,

‘workspace_name’: ‘apiorchestrationworkspacegsze’},

‘killed’: False,

‘created_at’: ‘2023-05-22T21:08:31.099447+00:00’,

‘updated_at’: ‘2023-05-22T21:08:36.585775+00:00’,

’last_runs’: [{‘run_id’: ‘96a7f85f-e30c-40b5-9185-0dee5bd1a15e’,

‘status’: ‘running’,

‘created_at’: ‘2023-05-22T21:08:33.112805+00:00’,

‘updated_at’: ‘2023-05-22T21:08:33.112805+00:00’}]}

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.

time.sleep(30)

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

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

datetime.datetime(2023, 5, 22, 21, 9, 32, 447321)

Get Tasks by Orchestration SHA

Tasks tied to the same orchestration are retrieved through the following request.

• REQUEST
  • POST /v1/api/task/get_tasks_by_orch_sha
• PARAMETERS
  • sha: The orchestrations SHA hash.
• RETURNS
  • ids: List[string] List of tasks by UUID.

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/get_tasks_by_orch_sha"

data = {
    "sha": orchestration_sha
}

response=requests.post(url, headers=headers, json=data).json()
display(response)

{'ids': ['2424a9a7-2331-42f6-bd84-90643386b130',
  'c868aa44-f7fe-4e3d-b11d-e1e6af3ec150',
  'a41fe4ae-b8a4-4e1f-a45a-114df64ae2bc']}

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. It returns the reverse chronological order of tasks runs listed by updated_at.

• REQUEST
  • POST /v1/api/task/list_task_runs
• PARAMETERS
  • task_id: The numerical identifier of the task.
  • status: Filters the task history by the status. If all, returns all statuses. Status values are:
    • running: The task has started.
    • failure: The task failed.
    • success: The task completed.
  • limit: The number of tasks runs to display.
• RETURNS
  • ids: List of task runs ids in UUID.

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/list_task_runs"

data = {
    "task_id": task_id
}

response=requests.post(url, headers=headers, json=data).json()
task_run_id = response[0]['run_id']
display(response)

[{'task': 'c868aa44-f7fe-4e3d-b11d-e1e6af3ec150',
  'run_id': '96a7f85f-e30c-40b5-9185-0dee5bd1a15e',
  'status': 'success',
  'created_at': '2023-05-22T21:08:33.112805+00:00',
  'updated_at': '2023-05-22T21:08:33.112805+00:00'}]

Get Task Run Logs

Logs for a task run are retrieved through the following process.

• REQUEST
  • POST /v1/api/task/get_logs_for_run
• PARAMETERS
  • id: The numerical identifier of the task run associated with the orchestration.
  • lines: The number of log lines to retrieve starting from the end of the log.
• RETURNS
  • logs: Array of log entries.

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/get_logs_for_run"

data = {
    "id": task_run_id
}

response=requests.post(url, headers=headers, json=data).json()
display(response)

{'logs': ["2023-05-22T21:09:17.683428502Z stdout F {'pipeline_name': 'apipipelinegsze', 'workspace_name': 'apiorchestrationworkspacegsze'}",
  '2023-05-22T21:09:17.683489102Z stdout F Getting the workspace apiorchestrationworkspacegsze',
  '2023-05-22T21:09:17.683497403Z stdout F Getting the pipeline apipipelinegsze',
  '2023-05-22T21:09:17.683504003Z stdout F Deploying the pipeline.',
  '2023-05-22T21:09:17.683510203Z stdout F Performing sample inference.',
  '2023-05-22T21:09:17.683516203Z stdout F                      time  ... check_failures',
  '2023-05-22T21:09:17.683521903Z stdout F 0 2023-05-22 21:08:37.779  ...              0',
  '2023-05-22T21:09:17.683527803Z stdout F ',
  '2023-05-22T21:09:17.683533603Z stdout F [1 rows x 4 columns]',
  '2023-05-22T21:09:17.683540103Z stdout F Undeploying the pipeline']}

Run Task Scheduled via API

The other method of using tasks is as a scheduled run through the Orchestration run_scheduled(name, schedule, timeout, json_args). This sets up a task to run on an regular schedule as defined by the schedule parameter in the cron service format. For example:

schedule={'42 * * * *'}

Runs on the 42nd minute of every hour.

The following schedule runs every day at 12 noon from February 1 to February 15 2024 - and then ends.

schedule={'0 0 12 1-15 2 2024'}

The Run Scheduled Task request is available at the following address:

/v1/api/task/run_scheduled

And takes the following parameters.

• name (String Required): The name to assign to the task.
• orch_id (String Required): The UUID orchestration ID to create the task from.
• workspace_id (Integer Required): The numberical identifier for the workspace.
• schedule (String Required): The schedule as a single string in cron format.
• timeout(Integer Optional): The timeout to complete the task in seconds.
• json (String Required): The arguments to pass to the task.

For our example, we will create a scheduled task to run every 5 minutes, display the inference results, then use the Orchestration kill task to keep the task from running any further.

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.

# retrieve the authorization token
headers = wl.auth.auth_header()

data = {
    "name": "scheduled api task",
    "workspace_id": workspace_id,
    "orch_id": orchestration_id,
    "schedule": "*/5 * * * *",
    "json": {
        "workspace_name": workspace_name,
        "pipeline_name": pipeline_name
    }
}

import datetime
task_start = datetime.datetime.now()

url=f"{APIURL}/v1/api/task/run_scheduled"

response=requests.post(url, headers=headers, json=data).json()
display(response)
scheduled_task_id = response['id']

{'id': '87c2c7c0-e17e-4c00-9407-bfc44d632910'}

# loop until the task is started

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/get_by_id"

data = {
    "id": scheduled_task_id
}

status = None

while status != 'started':
    response=requests.post(url, headers=headers, json=data).json()
    display(response)
    status = response['status']
    time.sleep(10)

{'name': 'scheduled api task',
 'id': '87c2c7c0-e17e-4c00-9407-bfc44d632910',
 'image': 'proxy.replicated.com/proxy/wallaroo/ghcr.io/wallaroolabs/arbex-orch-deploy',
 'image_tag': 'v2023.2.0-main-3271',
 'bind_secrets': ['minio'],
 'extra_env_vars': {'MINIO_URL': 'http://minio.wallaroo.svc.cluster.local:9000',
  'ORCH_OWNER_ID': 'a6857628-b0aa-451e-a0a0-bbc1d6eea6e0',
  'ORCH_SHA': 'd3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e',
  'TASK_DEBUG': 'false',
  'TASK_ID': '87c2c7c0-e17e-4c00-9407-bfc44d632910'},
 'auth_init': True,
 'workspace_id': 8,
 'schedule': '*/5 * * * *',
 'reap_threshold_secs': 900,
 'status': 'started',
 'input_data': {'pipeline_name': 'apipipelinegsze',
  'workspace_name': 'apiorchestrationworkspacegsze'},
 'killed': False,
 'created_at': '2023-05-22T21:10:22.43957+00:00',
 'updated_at': '2023-05-22T21:10:22.871615+00:00',
 'last_runs': []}

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/list_task_runs"

data = {
    "task_id": scheduled_task_id
}

# loop until we have a single run task
length = 0
while length == 0:
    response=requests.post(url, headers=headers, json=data).json()
    display(response)
    length = len(response)
    time.sleep(10)
task_run_id = response[0]['run_id']

[{'task': '87c2c7c0-e17e-4c00-9407-bfc44d632910',
  'run_id': 'ad3d427a-3643-4f10-9b94-116688b32355',
  'status': 'running',
  'created_at': '2023-05-22T21:15:02.148274+00:00',
  'updated_at': '2023-05-22T21:15:02.148274+00:00'}]

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/get_logs_for_run"

data = {
    "id": task_run_id
}

# loop until we get a log
response=requests.post(url, headers=headers, json=data).json()
display(response)

{'logs': ["2023-05-22T21:15:55.548754679Z stdout F {'pipeline_name': 'apipipelinegsze', 'workspace_name': 'apiorchestrationworkspacegsze'}",
  '2023-05-22T21:15:55.54880928Z stdout F Getting the workspace apiorchestrationworkspacegsze',
  '2023-05-22T21:15:55.54881708Z stdout F Getting the pipeline apipipelinegsze',
  '2023-05-22T21:15:55.54882328Z stdout F Deploying the pipeline.',
  '2023-05-22T21:15:55.54883088Z stdout F Performing sample inference.',
  '2023-05-22T21:15:55.54883628Z stdout F                      time  ... check_failures',
  '2023-05-22T21:15:55.54884248Z stdout F 0 2023-05-22 21:15:17.420  ...              0',
  '2023-05-22T21:15:55.54884788Z stdout F ',
  '2023-05-22T21:15:55.54885348Z stdout F [1 rows x 4 columns]',
  '2023-05-22T21:15:55.54885938Z stdout F Undeploying the pipeline']}

Cleanup

With the tutorial complete, we can undeploy the pipeline and return the resources back to the Wallaroo instance.

Kill Task via API

Started tasks are killed through the following request:

/v1/api/task/kill

And takes the following parameters.

• orch_id (String Required): The UUID orchestration ID to create the task from.

# retrieve the authorization token
headers = wl.auth.auth_header()

url=f"{APIURL}/v1/api/task/kill"

data = {
    "id": scheduled_task_id
}

response=requests.post(url, headers=headers, json=data).json()
display(response)

{'name': 'scheduled api task',
 'id': '87c2c7c0-e17e-4c00-9407-bfc44d632910',
 'image': 'proxy.replicated.com/proxy/wallaroo/ghcr.io/wallaroolabs/arbex-orch-deploy',
 'image_tag': 'v2023.2.0-main-3271',
 'bind_secrets': ['minio'],
 'extra_env_vars': {'MINIO_URL': 'http://minio.wallaroo.svc.cluster.local:9000',
  'ORCH_OWNER_ID': 'a6857628-b0aa-451e-a0a0-bbc1d6eea6e0',
  'ORCH_SHA': 'd3b93c9f280734106376e684792aa8b4285d527092fe87d89c74ec804f8e169e',
  'TASK_DEBUG': 'false',
  'TASK_ID': '87c2c7c0-e17e-4c00-9407-bfc44d632910'},
 'auth_init': True,
 'workspace_id': 8,
 'schedule': '*/5 * * * *',
 'reap_threshold_secs': 900,
 'status': 'pending_kill',
 'input_data': {'pipeline_name': 'apipipelinegsze',
  'workspace_name': 'apiorchestrationworkspacegsze'},
 'killed': False,
 'created_at': '2023-05-22T21:10:22.43957+00:00',
 'updated_at': '2023-05-22T21:10:22.871615+00:00',
 'last_runs': [{'run_id': 'ad3d427a-3643-4f10-9b94-116688b32355',
   'status': 'success',
   'created_at': '2023-05-22T21:15:02.148274+00:00',
   'updated_at': '2023-05-22T21:15:02.148274+00:00'}]}

Close Resources

With the tutorial complete, we’ll verify the pipeline is closed so the resources are assigned back to the Wallaroo instance.

pipeline.undeploy()

 ok

2 - Wallaroo Connection API with Google BigQuery Tutorial

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

Wallaroo Connection and ML Workload Orchestration with BigQuery House Price Model Tutorial

This tutorial provides a quick set of methods and examples regarding Wallaroo Connections. For full details, see the Wallaroo Documentation site.

Wallaroo provides Data Connections to organizations with a method of creating and managing automated tasks that can either be run on demand or a regular schedule.

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.

This tutorial will focus on using Google BigQuery as the data source.

Tutorial Goals

The tutorial will demonstrate the following:

1. Create a Wallaroo connection to retrieving information from a Google BigQuery source table.
2. Create a Wallaroo connection to store inference results into a Google BigQuery destination table.
3. Upload Wallaroo ML Workload Orchestration that supports BigQuery connections with the connection details.
4. Run the orchestration once as a Run Once Task and verify that the inference request succeeded and the inference results were saved to the external data store.
5. Schedule the orchestration as a Scheduled Task and verify that the inference request succeeded and the inference results were saved to the external data store.

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
  • pyarrow: PyArrow for Apache Arrow support
• The following Python libraries. These are not included in a Wallaroo instance’s JupyterHub service.
  • google-cloud-bigquery: Specifically for its support for Google BigQuery.
  • google-auth: Used to authenticate for bigquery.
  • db-dtypes: Converts the BigQuery results to Apache Arrow table or pandas DataFrame.

Tutorial Resources

• Models:
  • models/rf_model.onnx: A model that predicts house price values.
• Data:
  • data/xtest-1.df.json and data/xtest-1k.df.json: DataFrame JSON inference inputs with 1 input and 1,000 inputs.
  • data/xtest-1k.arrow: Apache Arrow inference inputs with 1 input and 1,000 inputs.
  • Sample inference inputs in CSV that can be imported into Google BigQuery.
    • data/xtest-1k.df.json: Random sample housing prices.
    • data/smallinputs.df.json: Sample housing prices that return results lower than $1.5 million.
    • data/biginputs.df.json: Sample housing prices that return results higher than $1.5 million.
  • SQL queries to create the inputs/outputs tables with schema.
    • ./resources/create_inputs_table.sql: Inputs table with schema.
    • ./resources/create_outputs_table.sql: Outputs table with schema.
    • ./resources/housrpricesga_inputs.avro: Avro container of inputs table.

Initial Steps

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

import time
import json

# for Big Query connections
from google.cloud import bigquery
from google.oauth2 import service_account
import db_dtypes

import requests

# used for unique connection names

import string
import random

suffix= ''.join(random.choice(string.ascii_lowercase) for i in range(4))

wallaroo.__version__

'2023.2.0rc3'

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(). If logging in externally, update the wallarooPrefix and wallarooSuffix variables with the proper DNS information. For more information on Wallaroo Client settings, see the Client Connection guide.

# Login through local Wallaroo instance

wl = wallaroo.Client()

API URL

The variable APIURL is used to specify the connection to the Wallaroo instance’s MLOps API URL, and is composed of the Wallaroo DNS prefix and suffix. For full details, see the Wallaroo API Connection Guide
.

The variables wallarooPrefix and wallarooSuffix variables will be used to derive the API url. For example, if the Wallaroo Prefix is doc-test. and the url is example.com, then the MLOps API URL would be doc-test.api.example.com/v1/api/{request}.

Note the . is part of the prefix. If there is no prefix, then wallarooPrefix = ""

Set the Wallaroo Prefix and Suffix in the code segment below based on your Wallaroo instance.

Variable Declaration

The following variables will be used for our big query testing.

We’ll use two connections:

• bigquery_input_connection: The connection that will draw inference input data from a BigQuery table.
• bigquery_output_connection: The connection that will upload inference results into a BigQuery table.

Not that for the connection arguments, we’ll retrieve the information from the files ./bigquery_service_account_input_key.json and ./bigquery_service_account_output_key.json that include the service account key file(SAK) information, as well as the dataset and table used.

# Setting variables for later steps

wallarooPrefix = "YOUR PREFIX."
wallarooSuffix = "YOUR SUFFIX"
APIURL = f"https://{wallarooPrefix}api.{wallarooSuffix}"

# Setting variables for later steps

workspace_name = 'bigqueryapiworkspace'
pipeline_name = 'bigqueryapipipeline'
model_name = 'bigqueryapimodel'
model_file_name = './models/rf_model.onnx'

bigquery_connection_input_name = f"bigqueryhouseapiinput{suffix}"
bigquery_connection_input_type = "BIGQUERY"
bigquery_connection_input_argument = json.load(open('./bigquery_service_account_input_key.json'))

bigquery_connection_output_name = f"bigqueryhouseapioutputs{suffix}"
bigquery_connection_output_type = "BIGQUERY"
bigquery_connection_output_argument = json.load(open('./bigquery_service_account_output_key.json'))

Helper Methods

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

# 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

def get_connection(name, connection_type, connection_arguments):
    try:
        connection = wl.get_connection(name)
    except RequiredAttributeMissing:
        connection =wl.create_connection(name, 
                  connection_type, 
                  connection_arguments)
    return connection

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)

workspace_id = workspace.id()

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(tensor_fields=["tensor"])
                        )

# Add the model as a pipeline step

pipeline.add_model_step(housing_model_control)

#deploy the pipeline
pipeline.deploy()

Connection Management via the Wallaroo MLOps API

The following steps will demonstration using the Wallaroo MLOps API to:

• Create the BigQuery connections
• Add the connections to the targeted workspace
• Use the connections for inference requests and uploading the results to a BigQuery dataset table.

Create Connections via API

We will create the data source connection via the Wallaroo api request:

/v1/api/connections/create

This takes the following parameters:

• name (String Required): The name of the connection.

• type (String Required): The user defined type of connection.

• details (String Required): User defined configuration details for the data connection. These can be {'username':'dataperson', 'password':'datapassword', 'port': 3339}, or {'token':'abcde123==', 'host':'example.com', 'port:1234'}, or other user defined combinations.

• IMPORTANT NOTE: Data connections names must be unique. Attempting to create a data connection with the same name as an existing data connection will result in an error.

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/connections/create"

# input connection
data = {
    'name': bigquery_connection_input_name,
    'type' : bigquery_connection_input_type,
    'details': bigquery_connection_input_argument
}

response=requests.post(url, headers=headers, json=data).json()
display(response)
# saved for later steps
connection_input_id = response['id']

{'id': '839779d7-d4b3-4e1c-953a-2f5ef55f1bb2'}

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/connections/create"

# output connection
data = {
    'name': bigquery_connection_output_name,
    'type' : bigquery_connection_output_type,
    'details': bigquery_connection_output_argument
}

response=requests.post(url, headers=headers, json=data).json()
display(response)
# saved for later steps
connection_output_id = response['id']

{'id': 'ef0c62e7-e59f-4c43-bdff-05ed0976cffa'}

Add Connections to Workspace via API

The connections will be added to the sample workspace with the MLOps API request:

/v1/api/connections/add_to_workspace

This takes the following parameters:

• workspace_id (String Required): The name of the connection.
• connection_id (String Required): The UUID connection ID

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/connections/add_to_workspace"

data = {
    'workspace_id': workspace_id,
    'connection_id': connection_input_id
}

response=requests.post(url, headers=headers, json=data)
display(response.json())

data = {
    'workspace_id': workspace_id,
    'connection_id': connection_output_id
}

response=requests.post(url, headers=headers, json=data)
display(response.json())

{'id': '3877da9d-e450-4b9a-85dd-7f68127b1313'}
{‘id’: ‘15f247ab-833f-468a-b7c5-9b50831052e6’}

Connect to Google BigQuery

With our connections set, we’ll now use them for an inference request through the following steps:

1. Retrieve the input data from a BigQuery request from the input connection details.
2. Perform the inference.
3. Upload the inference results into another BigQuery table from the output connection details.

Create Google Credentials

From our BigQuery request, we’ll create the credentials for our BigQuery connection.

We will use the MLOps API call:

/v1/api/connections/get

to retrieve the connection. This request takes the following parameters:

• name (String Required): The name of the connection.

# get the connection input details

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/connections/get"

data = {
    'name': bigquery_connection_input_name
}

connection_input_details=requests.post(url, headers=headers, json=data).json()['details']

# get the connection output details

# retrieve the authorization token
headers = wl.auth.auth_header()

url = f"{APIURL}/v1/api/connections/get"

data = {
    'name': bigquery_connection_output_name
}

connection_output_details=requests.post(url, headers=headers, json=data).json()['details']

# Set the bigquery credentials

bigquery_input_credentials = service_account.Credentials.from_service_account_info(
    connection_input_details)

bigquery_output_credentials = service_account.Credentials.from_service_account_info(
    connection_output_details)

Connect to Google BigQuery

We can now generate a client from our connection details, specifying the project that was included in the big_query_connection details.

bigqueryinputclient = bigquery.Client(
    credentials=bigquery_input_credentials, 
    project=connection_input_details['project_id']
)
bigqueryoutputclient = bigquery.Client(
    credentials=bigquery_output_credentials, 
    project=connection_output_details['project_id']
)

Query Data

Now we’ll create our query and retrieve information from out dataset and table as defined in the file bigquery_service_account_key.json. The table is expected to be in the format of the file ./data/xtest-1k.df.json.

inference_dataframe_input = bigqueryinputclient.query(
        f"""
        SELECT tensor
        FROM {connection_input_details['dataset']}.{connection_input_details['table']}"""
    ).to_dataframe()

inference_dataframe_input.head(5)

Sample Inference

With our data retrieved, we’ll perform an inference and display the results.

result = pipeline.infer(inference_dataframe_input)
display(result.head(5))

Upload the Results

With the query complete, we’ll upload the results back to the BigQuery dataset.

output_table = bigqueryoutputclient.get_table(f"{connection_output_details['dataset']}.{connection_output_details['table']}")

bigqueryoutputclient.insert_rows_from_dataframe(
    output_table, 
    dataframe=result.rename(columns={"in.tensor":"in_tensor", "out.variable":"out_variable"})
)

[[], []]

Verify the Upload

We can verify the upload by requesting the last few rows of the output table.

task_inference_results = bigqueryoutputclient.query(
        f"""
        SELECT *
        FROM {connection_output_details['dataset']}.{connection_output_details['table']}
        ORDER BY time DESC
        LIMIT 5
        """
    ).to_dataframe()

display(task_inference_results)

Cleanup

With the tutorial complete, we can undeploy the pipeline and return the resources back to the Wallaroo instance.

pipeline.undeploy()

3 - Wallaroo ML Workload Orchestration Simple Tutorial

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

Wallaroo Connection and ML Workload Orchestration Simple Tutorial

This tutorial provides a quick set of methods and examples regarding Wallaroo Connections and Wallaroo ML Workload Orchestration. For full details, see the Wallaroo Documentation site.

Wallaroo provides Data Connections and ML Workload Orchestrations to provide organizations with a method of creating and managing automated tasks that can either be run on demand or a regular schedule.

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 a Wallaroo connection to retrieving information from an external source.
2. Upload Wallaroo ML Workload Orchestration.
3. Run the orchestration once as a Run Once Task and verify that the information was saved the pipeline logs.
4. Schedule the orchestration as a Scheduled Task and verify that the information was saved to the pipeline logs.

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
  • pyarrow: PyArrow for Apache Arrow support

Initial Steps

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

import time

# 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)

'dtzw'

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'simpleorchestrationworkspace{suffix}'
pipeline_name = f'simpleorchestrationpipeline{suffix}'
model_name = f'simpleorchestrationmodel{suffix}'
model_file_name = './models/rf_model.onnx'

inference_connection_name = f'external_inference_connection{suffix}'
inference_connection_type = "HTTP"
inference_connection_argument = {'host':'https://github.com/WallarooLabs/Wallaroo_Tutorials/raw/main/wallaroo-testing-tutorials/houseprice-saga/data/xtest-1k.arrow'}

Helper Methods

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

# 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(tensor_fields=["tensor"])
                        )

# 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

Create Connections

We will create the data source connection via the Wallaroo client command create_connection.

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

• IMPORTANT NOTE: Data connections names must be unique. Attempting to create a data connection with the same name as an existing data connection will result in an error.

We’ll also create a data connection named inference_results_connection with our helper function get_connection that will either create or retrieve a connection if it already exists. From there we’ll create out connections:

• houseprice_arrow_table: An Apache Arrow file stored on GitHub that will be used for our inference input.

wl.create_connection(inference_connection_name, inference_connection_type, inference_connection_argument)

Get Connection by Name

The Wallaroo client method get_connection(name) retrieves the connection that matches the name parameter. We’ll retrieve our connection and store it as inference_source_connection.

inference_source_connection = wl.get_connection(name=inference_connection_name)
display(inference_source_connection)

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 both connections to our sample workspace, then list the connections available to the workspace to confirm.

workspace.add_connection(inference_connection_name)
workspace.list_connections()

Wallaroo ML Workload Orchestration 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. Use the inference_results_connection to open a HTTP Get connection to the inference data file and use it in an inference request in the deployed pipeline.
2. Submit the inference results to the location specified in the external_inference_connection.

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

Note that the orchestration assumes the pipeline is already deployed.

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/remote_inference.zip")

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

pending_packaging
pending_packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging

wl.list_orchestrations()

Upload Orchestration via File Object

Another method to upload the orchestration is as a file object. For that, we will open the zip file as a binary, then upload it using the bytes_buffer parameter to specify the file object, and the file_name to give it a new name.

zipfile = open("./remote_inference/remote_inference.zip", "rb").read()

wl.upload_orchestration(bytes_buffer=zipfile, file_name="inferencetest.zip", name="uploadedbytesdemo")

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.

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="simpletaskdemo", 
                              json_args={"workspace_name": workspace_name, 
                                         "pipeline_name": pipeline_name,
                                         "connection_name": inference_connection_name
                                            })

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)

'pending'
‘pending’
‘pending’

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 logs() method.

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.

time.sleep(60)

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

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

datetime.datetime(2023, 5, 23, 15, 28, 30, 718361)
Warning: Pipeline log size limit exceeded. Please request logs using export_logs

492 rows × 4 columns

Scheduled Run Task Example

The other method of using tasks is as a scheduled run through the Orchestration run_scheduled(name, schedule, timeout, json_args). This sets up a task to run on an regular schedule as defined by the schedule parameter in the cron service format. For example:

schedule={'42 * * * *'}

Runs on the 42nd minute of every hour.

The following schedule runs every day at 12 noon from February 1 to February 15 2024 - and then ends.

schedule={'0 0 12 1-15 2 2024'}

For our example, we will create a scheduled task to run every 5 minutes, display the inference results, then use the Orchestration kill task to keep the task from running any further.

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.

scheduled_task_start = datetime.datetime.now()

scheduled_task = orchestration.run_scheduled(name="simple_inference_schedule", 
                                             schedule="*/5 * * * *", 
                                             timeout=120, 
                                             json_args={"workspace_name": workspace_name, 
                                                        "pipeline_name": pipeline_name,
                                                        "connection_name": inference_connection_name
                                            })

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

'pending'

#wait 420 seconds to give the scheduled event time to finish
time.sleep(420)
scheduled_task_end = datetime.datetime.now()

pipeline.logs(start_datetime = scheduled_task_start, end_datetime = scheduled_task_end)

Warning: Pipeline log size limit exceeded. Please request logs using export_logs

492 rows × 4 columns

Kill Task

With our testing complete, we will kill the scheduled task so it will not run again. First we’ll show all the tasks to verify that our task is there, then issue it the kill command.

wl.list_tasks()

scheduled_task.kill()

<ArbexStatus.PENDING_KILL: 'pending_kill'>

wl.list_tasks()

Cleanup

With the tutorial complete, we can undeploy the pipeline and return the resources back to the Wallaroo instance.

pipeline.undeploy()

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

4 - Wallaroo ML Workload Orchestration Google BigQuery with House Price Model Tutorial

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

Wallaroo ML Workload Orchestration House Price Model Tutorial

This tutorial provides a quick set of methods and examples regarding Wallaroo Connections and Wallaroo ML Workload Orchestration. For full details, see the Wallaroo Documentation site.

Wallaroo provides Data Connections and ML Workload Orchestrations to provide organizations with a method of creating and managing automated tasks that can either be run on demand or a regular schedule.

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.

This tutorial will focus on using Google BigQuery as the data source.

Tutorial Goals

The tutorial will demonstrate the following:

1. Create a Wallaroo connection to retrieving information from a Google BigQuery source table.
2. Create a Wallaroo connection to store inference results into a Google BigQuery destination table.
3. Upload Wallaroo ML Workload Orchestration that supports BigQuery connections with the connection details.
4. Run the orchestration once as a Run Once Task and verify that the inference request succeeded and the inference results were saved to the external data store.
5. Schedule the orchestration as a Scheduled Task and verify that the inference request succeeded and the inference results were saved to the external data store.

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
  • pyarrow: PyArrow for Apache Arrow support
• The following Python libraries. These are not included in a Wallaroo instance’s JupyterHub service.
  • google-cloud-bigquery: Specifically for its support for Google BigQuery.
  • google-auth: Used to authenticate for bigquery.
  • db-dtypes: Converts the BigQuery results to Apache Arrow table or pandas DataFrame.

Tutorial Resources

• Models:
  • models/rf_model.onnx: A model that predicts house price values.
• Data:
  • data/xtest-1.df.json and data/xtest-1k.df.json: DataFrame JSON inference inputs with 1 input and 1,000 inputs.
  • data/xtest-1k.arrow: Apache Arrow inference inputs with 1 input and 1,000 inputs.
  • Sample inference inputs in CSV that can be imported into Google BigQuery.
    • data/xtest-1k.df.json: Random sample housing prices.
    • data/smallinputs.df.json: Sample housing prices that return results lower than $1.5 million.
    • data/biginputs.df.json: Sample housing prices that return results higher than $1.5 million.
  • SQL queries to create the inputs/outputs tables with schema.
    • ./resources/create_inputs_table.sql: Inputs table with schema.
    • ./resources/create_outputs_table.sql: Outputs table with schema.
    • ./resources/housrpricesga_inputs.avro: Avro container of inputs table.

Initial Steps

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

import time
import json

# for Big Query connections
from google.cloud import bigquery
from google.oauth2 import service_account
import db_dtypes

# used for unique connection names

import string
import random

suffix= ''.join(random.choice(string.ascii_lowercase) for i in range(4))

wallaroo.__version__

'2023.2.0+dfca0605e'

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()

Variable Declaration

The following variables will be used for our big query testing.

We’ll use two connections:

• bigquery_input_connection: The connection that will draw inference input data from a BigQuery table.
• bigquery_output_connection: The connection that will upload inference results into a BigQuery table.

Not that for the connection arguments, we’ll retrieve the information from the files ./bigquery_service_account_input_key.json and ./bigquery_service_account_output_key.json that include the service account key file(SAK) information, as well as the dataset and table used.

# Setting variables for later steps

workspace_name = f'bigqueryworkspace{suffix}'
pipeline_name = f'bigquerypipeline{suffix}'
model_name = f'bigquerymodel{suffix}'
model_file_name = './models/rf_model.onnx'

bigquery_connection_input_name = 'bigqueryhouseinputs{suffix}'
bigquery_connection_input_type = "BIGQUERY"
bigquery_connection_input_argument = json.load(open("./bigquery_service_account_input_key.json"))

bigquery_connection_output_name = 'bigqueryhouseoutputs{suffix}'
bigquery_connection_output_type = "BIGQUERY"
bigquery_connection_output_argument = json.load(open("./bigquery_service_account_output_key.json"))

Helper Methods

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

# 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(tensor_fields=["tensor"])
                        )

# Add the model as a pipeline step

pipeline.add_model_step(housing_model_control)

#deploy the pipeline to set the pipeline steps
pipeline.deploy()

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

Create Connections

We will create the data source connection via the Wallaroo client command create_connection.

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

• IMPORTANT NOTE: Data connections names must be unique. Attempting to create a data connection with the same name as an existing data connection will result in an error.

connection_input = wl.create_connection(bigquery_connection_input_name, bigquery_connection_input_type, bigquery_connection_input_argument)
connection_output = wl.create_connection(bigquery_connection_output_name, bigquery_connection_output_type, bigquery_connection_output_argument)
wl.list_connections()

Get Connection by Name

The Wallaroo client method get_connection(name) retrieves the connection that matches the name parameter. We’ll retrieve our connection and store it as inference_source_connection.

big_query_input_connection = wl.get_connection(name=bigquery_connection_input_name)
big_query_output_connection = wl.get_connection(name=bigquery_connection_output_name)
display(big_query_input_connection)
display(big_query_output_connection)

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 both connections to our sample workspace, then list the connections available to the workspace to confirm.

workspace.add_connection(bigquery_connection_input_name)
workspace.add_connection(bigquery_connection_output_name)

workspace.list_connections()

Big Query Connection Inference Example

We can test the BigQuery connection with a simple inference to our deployed pipeline. We’ll request the data, format the table into a pandas DataFrame, then submit it for an inference request.

Create Google Credentials

From our BigQuery request, we’ll create the credentials for our BigQuery connection.

bigquery_input_credentials = service_account.Credentials.from_service_account_info(
    big_query_input_connection.details())

bigquery_output_credentials = service_account.Credentials.from_service_account_info(
    big_query_output_connection.details())

Connect to Google BigQuery

We can now generate a client from our connection details, specifying the project that was included in the big_query_connection details.

bigqueryinputclient = bigquery.Client(
    credentials=bigquery_input_credentials, 
    project=big_query_input_connection.details()['project_id']
)
bigqueryoutputclient = bigquery.Client(
    credentials=bigquery_output_credentials, 
    project=big_query_output_connection.details()['project_id']
)

Query Data

Now we’ll create our query and retrieve information from out dataset and table as defined in the file bigquery_service_account_key.json. The table is expected to be in the format of the file ./data/xtest-1k.df.json.

inference_dataframe_input = bigqueryinputclient.query(
        f"""
        SELECT tensor
        FROM {big_query_input_connection.details()['dataset']}.{big_query_input_connection.details()['table']}"""
    ).to_dataframe()

inference_dataframe_input.head(5)

Sample Inference

With our data retrieved, we’ll perform an inference and display the results.

result = pipeline.infer(inference_dataframe_input)
display(result.head(5))

Upload the Results

With the query complete, we’ll upload the results back to the BigQuery dataset.

output_table = bigqueryoutputclient.get_table(f"{big_query_output_connection.details()['dataset']}.{big_query_output_connection.details()['table']}")

bigqueryoutputclient.insert_rows_from_dataframe(
    output_table, 
    dataframe=result.rename(columns={"in.tensor":"in_tensor", "out.variable":"out_variable"})
)

[[], []]

Wallaroo ML Workload Orchestration 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. Use the bigquery_remote_inference to open a connection to the input and output tables.
2. Deploy the pipeline.
3. Perform an inference with the input data.
4. Save the inference results to the output table.
5. Undeploy the pipeline.

This sample script is stored in bigquery_remote_inference/main.py with an requirements.txt file having the specific libraries for the Google BigQuery connection., and packaged into the orchestration as ./bigquery_remote_inference/bigquery_remote_inference.zip. We’ll display the steps in uploading the orchestration to the Wallaroo instance.

Note that the orchestration assumes the pipeline is already deployed.

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 ./bigquery_remote_inference/bigquery_remote_inference.zip will be uploaded and saved to the variable orchestration. Then we will loop until the uploaded orchestration’s status displays ready.

orchestration = wl.upload_orchestration(path="./bigquery_remote_inference/bigquery_remote_inference.zip")

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

pending_packaging
pending_packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging
packaging

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.

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.

We’ll display the last 5 rows of our BigQuery output table, then start the task that will perform the same inference we did above.

task_inference_results = bigqueryoutputclient.query(
        f"""
        SELECT *
        FROM {big_query_output_connection.details()['dataset']}.{big_query_output_connection.details()['table']}
        ORDER BY time DESC
        LIMIT 5
        """
    ).to_dataframe()

display(task_inference_results)

# Example: run once

import datetime
task_start = datetime.datetime.now()

task = orchestration.run_once(name="big query single run", json_args={})
task

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)

'pending'
‘pending’

wl.list_tasks()

Task Results

We can view the inferences from our logs and verify that new entries were added from our task. We’ll query the last 5 rows of our inference output table after a wait of 60 seconds.

time.sleep(60)

task_inference_results = bigqueryoutputclient.query(
        f"""
        SELECT *
        FROM {big_query_output_connection.details()['dataset']}.{big_query_output_connection.details()['table']}
        ORDER BY time DESC LIMIT 5"""
    ).to_dataframe()

display(task_inference_results)

Scheduled Run Task Example

The other method of using tasks is as a scheduled run through the Orchestration run_scheduled(name, schedule, timeout, json_args). This sets up a task to run on an regular schedule as defined by the schedule parameter in the cron service format. For example:

schedule={'42 * * * *'}

Runs on the 42nd minute of every hour.

The following schedule runs every day at 12 noon from February 1 to February 15 2024 - and then ends.

schedule={'0 0 12 1-15 2 2024'}

For our example, we will create a scheduled task to run every 5 minutes, display the inference results, then use the Orchestration kill task to keep the task from running any further.

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_inference_results = bigqueryoutputclient.query(
        f"""
        SELECT *
        FROM {big_query_output_connection.details()['dataset']}.{big_query_output_connection.details()['table']}
        ORDER BY time DESC LIMIT 5"""
    ).to_dataframe()

display(task_inference_results.tail(5))

scheduled_task = orchestration.run_scheduled(name="simple_inference_schedule", schedule="*/5 * * * *", timeout=120, json_args={})

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

'pending'

#wait 420 seconds to give the scheduled event time to finish
time.sleep(420)
task_inference_results = bigqueryoutputclient.query(
        f"""
        SELECT *
        FROM {big_query_output_connection.details()['dataset']}.{big_query_output_connection.details()['table']}
        ORDER BY time DESC LIMIT 5"""
    ).to_dataframe()

display(task_inference_results.tail(5))

Kill Task

With our testing complete, we will kill the scheduled task so it will not run again. First we’ll show all the tasks to verify that our task is there, then issue it the kill command.

scheduled_task.kill()

<ArbexStatus.PENDING_KILL: 'pending_kill'>

Cleanup

With the tutorial complete, we can undeploy the pipeline and return the resources back to the Wallaroo instance.

pipeline.undeploy()

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

5 - Wallaroo ML Workload Orchestration Google BigQuery with Statsmodel Forecast Tutorial

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

Wallaroo ML Workload Orchestrations and BigQuery Connections with Statsmodel Tutorial

This tutorial provides a quick set of methods and examples regarding Wallaroo Connections and Wallaroo ML Workload Orchestration. For full details, see the Wallaroo Documentation site.

Wallaroo provides Data Connections and ML Workload Orchestrations to provide organizations with a method of creating and managing automated tasks that can either be run on demand or a regular schedule.

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.

This tutorial will focus on using Google BigQuery as the data source for supplying the inference data to perform inferences through a Statsmodel ML model.

Tutorial Goals

The tutorial will demonstrate the following:

1. Create a Wallaroo connection to retrieving information from a Google BigQuery source table.
2. Create a Wallaroo connection to store inference results into a Google BigQuery destination table.
3. Upload Wallaroo ML Workload Orchestration that supports BigQuery connections with the connection details.
4. Run the orchestration once as a Run Once Task and verify that the inference request succeeded and the inference results were saved to the external data store.
5. Schedule the orchestration as a Scheduled Task and verify that the inference request succeeded and the inference results were saved to the external data store.

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
  • pyarrow: PyArrow for Apache Arrow support
• The following Python libraries. These are not included in a Wallaroo instance’s JupyterHub service.
  • google-cloud-bigquery: Specifically for its support for Google BigQuery.
  • google-auth: Used to authenticate for bigquery.
  • db-dtypes: Converts the BigQuery results to Apache Arrow table or pandas DataFrame.

Tutorial Resources

• Models:
  • models/bike_day_model.pkl: The statsmodel model predicts how many bikes will be rented on each of the next 7 days, based on the previous 7 days’ bike rentals, temperature, and wind speed. This model only accepts shapes (7,4) - 7 rows (representing the last 7 days) and 4 columns (representing the fields temp, holiday, workingday, windspeed). Additional files to support this example are:
  • infer.py: The inference script that is part of the statsmodel.
• Data:
  • data/day.csv: Data used to train the sample statsmodel example.
  • data/bike_day_eval.json: Data used for inferences. This will be transferred to a BigQuery table as shown in the demonstration.
• Resources:
  • resources/bigquery_service_account_input_key.json: Example service key to authenticate to a Google BigQuery project with the dataset and table used for the inference data.
  • resources/bigquery_service_account_output_key.json: Example service key to authenticate to a Google BigQuery project with the dataset and table used for the inference result data.
  • resources/statsmodel_forecast_inputs.sql: SQL script to create the inputs table schema, populated with the values from ./data/day.csv.
  • resources/statsmodel_forecast_outputs.sql: SQL script to create the outputs table schema.

Initial Steps

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

import time
import json

# for Big Query connections
from google.cloud import bigquery
from google.oauth2 import service_account
import db_dtypes

import string
import random

suffix= ''.join(random.choice(string.ascii_lowercase) for i in range(4))

wallaroo.__version__

'2023.2.0+dfca0605e'

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()

Variable Declaration

The following variables will be used for our big query testing.

We’ll use two connections:

• bigquery_input_connection: The connection that will draw inference input data from a BigQuery table.
• bigquery_output_connection: The connection that will upload inference results into a BigQuery table.

Not that for the connection arguments, we’ll retrieve the information from the files ./bigquery_service_account_input_key.json and ./bigquery_service_account_output_key.json that include the service account key file(SAK) information, as well as the dataset and table used.

# Setting variables for later steps

workspace_name = f'bigquerystatsmodelworkspace{suffix}'
pipeline_name = f'bigquerystatsmodelpipeline{suffix}'
model_name = f'bigquerystatsmodelmodel{suffix}'
model_file_name = './models/bike_day_model.pkl'

bigquery_connection_input_name = f'bigqueryforecastinputs{suffix}'
bigquery_connection_input_type = "BIGQUERY"
bigquery_connection_input_argument = json.load(open('./resources/bigquery_service_account_input_key.json'))

bigquery_connection_output_name = f'bigqueryforecastoutputs{suffix}'
bigquery_connection_output_type = "BIGQUERY"
bigquery_connection_output_argument = json.load(open('./resources/bigquery_service_account_output_key.json'))

Helper Methods

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

# 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

bike_day_model = (wl.upload_model(model_name, 
                                  model_file_name, 
                                  framework=wallaroo.framework.Framework.ONNX)
                                  .configure(runtime="python", tensor_fields=["tensor"])
                )

# Add the model as a pipeline step

pipeline.add_model_step(bike_day_model)

#deploy the pipeline
pipeline.deploy()

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

Sample Inferences

We’ll perform some quick sample inferences with the local file data to verity the pipeline deployed and is ready for inferences. Once done, we’ll undeploy the pipeline.

## perform inferences

results = pipeline.infer_from_file('./data/bike_day_eval.json', data_format="custom-json")
print(results)

[{'forecast': [1882.3784555157672, 2130.607915701861, 2340.84005381799, 2895.754978552066, 2163.657515565616, 1509.1792126509536, 2431.1838923957016]}]

Create Connections

We will create the data source connection via the Wallaroo client command create_connection.

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

• IMPORTANT NOTE: Data connections names must be unique. Attempting to create a data connection with the same name as an existing data connection will result in an error.

See the statsmodel_forecast_inputs and statsmodel_forecast_outputs details listed above for the table schema used for our example.

connection_input = wl.create_connection(bigquery_connection_input_name, bigquery_connection_input_type, bigquery_connection_input_argument)
connection_output = wl.create_connection(bigquery_connection_output_name, bigquery_connection_output_type, bigquery_connection_output_argument)

display(connection_input)
display(connection_output)

Get Connection by Name

The Wallaroo client method get_connection(name) retrieves the connection that matches the name parameter. We’ll retrieve our connection and store it as inference_source_connection.

big_query_input_connection = wl.get_connection(name=bigquery_connection_input_name)
big_query_output_connection = wl.get_connection(name=bigquery_connection_output_name)
display(big_query_input_connection)
display(big_query_output_connection)

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 both connections to our sample workspace, then list the connections available to the workspace to confirm.

workspace.add_connection(bigquery_connection_input_name)
workspace.add_connection(bigquery_connection_output_name)

workspace.list_connections()

Big Query Connection Inference Example

We can test the BigQuery connection with a simple inference to our deployed pipeline. We’ll request the data, format the table into a pandas DataFrame, then submit it for an inference request.

Create Google Credentials