Wallaroo provides ML Workload Orchestrations and Tasks to automate processes in a Wallaroo instance. For example:

• Deploy a pipeline, retrieve data through a data connector, submit the data for inferences, undeploy the pipeline
• Replace a model with a new version
• Retrieve shadow deployed inference results and submit them to a database

Orchestration Flow

ML Workload Orchestration flow works within 3 tiers:

One example may be of making donuts.

• The ML Workload Orchestration is the recipe.
• The Task is the order to make the donuts. It might be Run Once, so only one set of donuts are made, or Run Scheduled, so donuts are made every 2nd Sunday at 6 AM. If Run Scheduled, the donuts are made every time the schedule hits until the order is cancelled (aka killed).
• The Task Run are the donuts with their own receipt of creation (logs, etc).

Orchestration Requirements

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

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

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

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.1 Wallaroo instance, use the Wallaroo SDK version 2023.2.1.
• 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 orchestration task.
  • wallaroo.task_args(): Returns a Dict of invocation-specific arguments passed to the run_ calls.
• Orchestrations will be run in the same way as running within the Wallaroo JupyterHub service, from the version of Python libraries (unless specifically overridden by the requirements.txt setting, which is not recommended), and running in the virtualized directory /home/jovyan/.

Inference from Zero Scaled Deployments

For deployments that autoscale from 0 replicas, replica_autoscale_min_max is set with minimum=0 and replicas scale down to zero when there is no utilization based on the autoscale parameters. When a new inference request is made, the first replica is scaled up. Once the first replica is ready, inference requests proceed as normal.

When inferencing in this scenario, a timeout may occur waiting for the first replica to spool up. To handle situations where an autoscale deployment scales down to zero replicas, the following code example provides a way to “wake up” the pipeline with an inference request which may use mock or real data. Once the first replica is fully spooled up, inference requests proceed at full speed.

Once deployed, we check the pipeline’s deployment status to verify it is running. If the pipeline is still scaling, the process waits 10 seconds to allow it to finish scaling before attempting the initial inference again. Once an inference completes successfully, the inferences proceed as normal.

# verify deployment has the status `Running`
while pipeline.status()["status"] != 'Running':
    try:
        # attempt the inference
        pipeline.infer(dataframe)
    except:
        # if an exception is thrown, pass it
        pass
    # wait 10 seconds before attempting the inference again
    time.sleep(10)

# when the inference passes successfully, continue with other inferences as normal
pipeline.infer(dataframe2)
pipeline.infer(dataframe3)

Orchestration Code Samples

The following demonstrates using the wallaroo.in_task() and wallaroo.task_args() methods within an Orchestration. This sample code uses wallaroo.in_task() to verify whether or not the script is running as a Wallaroo Task. If true, it will gather the wallaroo.task_args() and use them to set the workspace and pipeline. If False, then it sets the pipeline and workspace manually.

# get the arguments
wl = wallaroo.Client()

# if true, get the arguments passed to the task
if wl.in_task():
  arguments = wl.task_args()
  
  # arguments is a key/value pair, set the workspace and pipeline name
  workspace_name = arguments['workspace_name']
  pipeline_name = arguments['pipeline_name']
  
# False:  We're not in a Task, so set the pipeline manually
else:
  workspace_name="bigqueryworkspace"
  pipeline_name="bigquerypipeline"

Manage ML Workload Orchestrations via the Wallaroo SDK

The following methods are provided for creating and listing orchestrations.

Orchestration Methods

Create Orchestration

An orchestration is created through the wallaroo.client.upload_orchestration(path) with the following parameters.

For the uploads, either the path to the .zip file is required, or bytes_buffer with name are required. path can not be used with bytes_buffer and name, and vice versa.

List Orchestrations

All orchestrations for a Wallaroo instances are listed via the Wallaroo Client list_orchestrations() method. The orchestration list returned is based on the workspaces the user is a member of and the filtered parameters, in reverse chronological order.

Admin users have unrestricted access to all workspaces. For more details, see Wallaroo Enterprise User Management.

List Orchestrations Parameters

List Orchestrations Returns

Returns a List of Orchestrations based on the filtered parameters. Each Orchestration includes the following fields.

List Orchestration Errors

The errors for this method include:

• If the parameter workspace_id is not an integer.
• If the parameter workspace_name is not a String.

List Orchestrations Examples

List all orchestrations available across workspaces, then filtered by workspace id, then filtered by workspace name.

wl.list_orchestrations()

wl.list_orchestrations(workspace_id=30)

wl.list_orchestrations(workspace_name="simpleorchestrationworkspace")

Task Methods

Tasks are the implementation of an orchestration. Think of the orchestration as the instructions to follow, and the Task is the unit actually doing it.

Tasks are set at the workspace level.

Create Tasks

Tasks are created from an orchestration through the following methods.

Tasks have the following parameters.

Run Task Once

Temporary Run tasks are created from the Orchestration run_once(name, json_args, timeout) with the following parameters.

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

Run Task Scheduled

A task can be scheduled via the Orchestration run_scheduled method.

Scheduled tasks are run every time the schedule period is met. This uses the same settings as the cron utility.

Scheduled tasks include the following parameters.

The schedule uses the same method as the cron service. For example, the following schedule:

schedule={'42 * * * *'}

Runs on the 42nd minute of every hour. The following schedule:

schedule={'00 1 * * 0'}

Indicates “At 1:00 AM on Sunday.”

For a shortcut in creating cron formatted schedules, see sites such as the Cron expression generator by Cronhub.

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
                                            })
task_scheduled

Run Task Continuously

Run Continuously Tasks are launched via the wallaroo.Orchestration.run_continuously method. These tasks continue to run until the kill task command is issued. Typical use cases include:

• Polling a data store for new records, and use those records for a new inference request.
• Check inference results for anomalies on a regular basis and send a message to interested parties.

The main difference between Run Scheduled and Run Continuously Tasks is that Run Scheduled generates a new Task Run each time a scheduled pattern is met, with each Task Run having its own Task Run Logs.

A Run Continuously Task is not bound by schedule, but on Task Run execution and completion. When a Run Continuously Task is launched:

• A new Task Run is generated.
• The Task Run will execute until either successful completion or failure (for example: an uncaught exception causes the script to terminate).
• Within one minute of the script’s completion or termination, a new Task Run is generated. Compare this with a Scheduled Task, which only generates a new Task Run based on the schedule.
  • Each separate Task Run has its own logs.
• This cycle continues until the the kill task command is issued to the Run Continuously Task.

This means that if a Run Continuously Task has an infinite loop that continues to run without terminating (for example: polling a database for new records every 5 minutes), only one Task Run is generated until the kill command is issued.

The following are two examples of an Orchestration and how they Run Continuously Tasks behave.

The following script executes and completes without any loops, and each time it completes, a new Task Run is generated with its own Task Run Logs.

import wallaroo
import pandas as pd
import pyarrow as pa
import requests
import time

# create a Wallaroo client
wl = wallaroo.Client()

# get the arguments
arguments = wl.task_args()

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

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

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

# print the parameters
print(workspace_name)
print(pipeline_name)
print(connection_name)

The following script will loop forever, generating only one Task Run, since the script never completes.

import wallaroo
import pandas as pd
import pyarrow as pa
import requests
import time

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="simpleorchestrationworkspace"

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

if "connection_name" in arguments:
    connection_name = arguments['connection_name']
else:
    connection_name = "external_inference_connection"
    
while True:
    print(workspace_name)
    print(pipeline_name)
    print(connection_name)
    # wait 60 seconds then print again
    time.sleep(60)

A Continuous Task is created via the Orchestration run_continuous method and includes the following parameters.

The following example generates a new Run Continuously Task and passes a set of arguments for the Task via the json_args parameter.

task_continuously = orchestration.run_continuously(name=sample_task_name, 
                              json_args={"workspace_name": workspace_name, 
                                         "pipeline_name": pipeline_name,
                                         "connection_name": connection_name
                                            })

display(task_continuously)                             

Run Continuously Tasks Best Practices

The following best practices are recommended for Run Continuously Tasks.

• For faster Task Run Log updates, flush Python Buffers: Outputs from commands including print typically do not appear in the Task Run Logs until the Python buffers are flushed, usually when the Python script completes. This allows the script to perform at peak performance. To have outputs show up in the Task Run logs faster for debugging or other purposes:
  • Flushing the output immediately. For example, print('sample', flush=True).
  • For helm based installations of Wallaroo, shorten the dbcleaner.schedule so the outputs are flushed sooner.
• Set log storage: By default, Task Run Logs are preserved for running Run Continuously Tasks for 30 days. For helm based installation of Wallaroo, this is adjusted through the dbcleaner.maxAgeDays setting.

Tasks with Inference from Zero Scaled Deployments

For deployments that autoscale from 0 replicas, replica_autoscale_min_max is set with minimum=0 and replicas scale down to zero when there is no utilization for 60 minutes. When a new inference request is made, the first replica is scaled up. Once the first replica is ready, inference requests proceed as normal.

When inferencing in this scenario, a timeout may occur waiting for the first replica to finish spooling up. To handle situations where an autoscale deployment scales down to zero replicas, the following code example provides a way to “wake up” the pipeline with an inference request which may use mock or real data. Once the first replica is fully spooled up, inference requests proceed at full speed.

The following deploys a pipeline with 4 cpus and 3 GB RAM per replica, with the autoscale set between 0 and 5 replicas.

Once deployed, we check the pipeline’s deployment status to verify it is running. If the pipeline is still scaling, the process waits 10 seconds to allow it to finish scaling. Once an inference completes successfully, the inferences proceed as normal.

# deployment configuration with autoscaling between 0 and 5 replicas
deployment_configuration = wallaroo.DeploymentConfigBuilder()
    .autoscale_cpu_utilization(75)
    .cpus(4)
    .memory('3Gi')
    .replica_autoscale_min_max(minimum=0, maximum=5)
    .build()

# deployment with the deployment configuration
pipeline.deploy(deployment_configuration)

# verify deployment has the status `Running`
while pipeline.status()["status"] != 'Running':
    try:
        # attempt the inference
        pipeline.infer(dataframe)
    except:
        # if an exception is thrown, pass it
        pass
    # wait 10 seconds before attempting the inference again
    time.sleep(10)
# when the inference passes successfully, continue with other inferences as normal
pipeline.infer(dataframe2)
pipeline.infer(dataframe3)
...

List Tasks

The list of tasks in the Wallaroo instance is retrieves through the Wallaroo Client list_tasks() method. The task list returned is based on the workspaces the user is a member of and the filtered parameters, in reverse chronological order.

Admin users have unrestricted access to all workspaces. For more details, see Wallaroo Enterprise User Management.

List Tasks Parameters

List Tasks Returns

A List of tasks based on the filtered parameters with the following fields per task:

List Tasks Errors

The errors for this method include:

• If the parameter workspace_id is not an integer.
• If the parameter workspace_name is not a String.

List Tasks Examples

Lists all tasks available across workspaces, then filtered by workspace id, then filtered by workspace name.

wl.list_tasks()

wl.list_tasks(workspace_id=30)

wl.list_tasks(workspace_name="simpleorchestrationworkspace")

Get Task Status

The status of a task is retrieved through the Task status() method and returns the following.

display(task2.status())
'started'

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.

Killed tasks have the following effects:

• Any active Task Runs are halted.
• All Task Runs Logs generated from any Task Runs created by the Task are deleted.

Note that a Task set to Run Scheduled will generate a new Task Run each time the schedule parameters are met until the Task is killed. A Task set to Run Once will generate only one Task Run, and a Run Continuously Task will only generate new Task Runs if the current Task Run either terminates or completes successfully. In all three cases, any active Task Runs are terminated and their associated Task Run Logs are deleted.

task2.kill()

<ArbexStatus.PENDING_KILL: 'pending_kill'>

Task Runs

Task Runs are generated from a Task.

• Run Once Tasks generated only one Task Run whether it completes successfully or terminates early.
• Run Once Tasks generate a new Task Run each time the schedule parameters are met, with each Task Run having its own results and logs. New Task Runs are generated whether the previous Task Runs completed or not.
• Run Continuously Tasks generate a new Task Run within one minute of the most recent Task Run completes either successfully or terminates early.

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 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<item: float> not null
2023-22-May 19:59:29   child 0, item: float
2023-22-May 19:59:29 out.variable: list<inner: float not null> not null
2023-22-May 19:59:29 anomaly.count: 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]]]</code></pre>