Wallaroo Observability

• 1: Wallaroo Edge Computer Vision Observability
• 2: Wallaroo Model Observability: Anomaly Detection with CCFraud
• 3: Wallaroo Model Observability: Anomaly Detection with House Price Prediction
• 4: Wallaroo Edge Observability with Classification Financial Models
• 5: Wallaroo Assays Model Insights Tutorial
• 6: Computer Vision Pipeline Logs MLOps API Tutorial
• 7: Wallaroo Edge Observability with Classification Financial Models through API
• 8: Pipeline Logs MLOps API Tutorial
• 9: Pipeline Logs Tutorial
• 10: Wallaroo Edge Observability with Wallaroo Assays
• 11: House Price Testing Life Cycle

1 - Wallaroo Edge Computer Vision Observability

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

Computer Vision for Object Detection in Retail

The following tutorial demonstrates using Wallaroo for observability of edge deployments of computer vision models.

Introduction

This tutorial focuses on the resnet50 computer vision model. By default, this provides the following outputs from receiving an image converted to tensor values:

• boxes: The bounding boxes for detected objects.
• classes: The class of the detected object (bottle, coat, person, etc).
• confidences: The confidence the model has that the detected model is the class.

For this demonstration, the model is modified with Wallaroo Bring Your Own Predict (BYOP) to add two additional fields:

• avg_px_intensity: the average pixel intensity checks the input to determine the average value of the Red/Green/Blue pixel values. This is used as a benchmark to determine if the two images are significantly different. For example, an all white photo would have an avg_px_intensity of 1, while an all blank photo would have a value of 0.
• avg_confidence: The average confidence of all detected objects.

This demonstration will use avg_confidence to demonstrate observability in our edge deployed computer vision model.

Prerequisites

• A Wallaroo Ops instance 2023.4 and above with [edge deployment enabled](Edge Deployment Registry Guide).
• An x64 edge device with Docker installed, recommended with at least 8 cores.

In order for the wallaroo tutorial notebooks to run properly, the videos directory must contain these models in the models directory.

To download the Wallaroo Computer Vision models, use the following link:

https://storage.googleapis.com/wallaroo-public-data/cv-demo-models/cv-retail-models.zip

Unzip the contents into the directory models.

The following models are required to run the tutorial:

onnx==1.12.0
onnxruntime==1.12.1
torchvision
torch
matplotlib==3.5.0
opencv-python
imutils
pytz
ipywidgets

To run this tutorial outside of a Wallaroo Ops center, the Wallaroo SDK is available and is installed via pip with:

pip install wallaroo==2023.4.1

References

• Wallaroo SDK Essentials Guide: Pipeline Edge Publication

Steps

Import Libraries

The following libraries are used to execute this tutorial. The utils.py provides additional helper methods for rendering the images into tensor fields and other useful tasks.

# preload needed libraries 

import wallaroo
from wallaroo.object import EntityNotFoundError
from wallaroo.framework import Framework
from IPython.display import display
from IPython.display import Image
import pandas as pd
import json
import datetime
import time
import cv2
import matplotlib.pyplot as plt
import string
import random
import pyarrow as pa
import sys
import asyncio
import numpy as np

import utils
pd.set_option('display.max_colwidth', None)

import datetime

# api based inference request
import requests

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.

The option request_timeout provides additional time for the Wallaroo model upload process to complete.

wl = wallaroo.Client(request_timeout=600)

Create Workspace

We will create a workspace to manage our pipeline and models. The following variables will set the name of our sample workspace then set it as the current workspace.

Workspace names must be unique. The following helper function will either create a new workspace, or retrieve an existing one with the same name. Verify that a pre-existing workspace has been shared with the targeted user.

Set the variables workspace_name to ensure a unique workspace name if required.

The workspace will then be set as the Current Workspace. Model uploads and

• References
  • Wallaroo SDK Essentials Guide: Workspace Management

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

workspace_name = "cv-retail-edge-observability"
model_name = "resnet-with-intensity"
model_file_name = "./models/model-with-pixel-intensity.zip"
pipeline_name = "retail-inv-tracker-edge-obs"

workspace = get_workspace(workspace_name, wl)
wl.set_current_workspace(workspace)

{'name': 'cv-retail-edge-observability', 'id': 8, 'archived': False, 'created_by': 'cc3619bb-2cec-4a44-9333-55a0dc6b3997', 'created_at': '2024-01-10T17:37:38.659309+00:00', 'models': [], 'pipelines': []}

Upload Model

The model is uploaded as a BYOP model, where the model, Python script and other artifacts are included in a .zip file. This requires the input and output schemas for the model specified in Apache Arrow Schema format.

• References
  • Wallaroo SDK Essentials Guide: Model Uploads and Registrations: Arbitrary Python

input_schema = pa.schema([
    pa.field('tensor', pa.list_(
        pa.list_(
            pa.list_(
                pa.float32(), # images are normalized
                list_size=640
            ),
            list_size=480
        ),
        list_size=3
    )),
])

output_schema = pa.schema([
    pa.field('boxes', pa.list_(pa.list_(pa.float32(), list_size=4))),
    pa.field('classes', pa.list_(pa.int64())),
    pa.field('confidences', pa.list_(pa.float32())),
    pa.field('avg_px_intensity', pa.list_(pa.float32())),
    pa.field('avg_confidence', pa.list_(pa.float32())),
])

model = wl.upload_model(model_name, 
                        model_file_name, 
                        framework=Framework.CUSTOM,
                        input_schema=input_schema, 
                        output_schema=output_schema)

Waiting for model loading - this will take up to 10.0min.
Model is pending loading to a container runtime......
Model is attempting loading to a container runtime.............successful

Ready

Deploy Pipeline

Next we configure the hardware we want to use for deployment. If we plan on eventually deploying to edge, this is a good way to simulate edge hardware conditions. The BYOP model is deployed as a Wallaroo Containerized Runtime, so the hardware allocation is performed through the sidekick options.

• References
  • Wallaroo SDK Essentials Guide: Pipeline Deployment Configuration

deployment_config = wallaroo.DeploymentConfigBuilder() \
    .replica_count(1) \
    .cpus(1) \
    .memory("2Gi") \
    .sidekick_cpus(model, 1) \
    .sidekick_memory(model, '6Gi') \
    .build()

We create the pipeline with the wallaroo.client.build_pipeline method, and assign our model as a model pipeline step. Once complete, we will deploy the pipeline to allocate resources from the Kuberntes cluster hosting the Wallaroo Ops to the pipeline.

pipeline = wl.build_pipeline(pipeline_name)
pipeline.clear()
pipeline.add_model_step(model)
pipeline.deploy(deployment_config = deployment_config)

Monitoring for Model Drift

For this example, we want to track the average confidence of object predictions and get alerted if we see a drop in confidence.

We will convert a set of images to pandas DataFrames with the images converted to tensor values. The first set of images baseline_images as well formed images. The second set blurred_images are the same photos intentionally blurred for our observability demonstration.

baseline_images = [
    "./data/images/input/example/dairy_bottles.png",
    "./data/images/input/example/dairy_products.png",
    "./data/images/input/example/product_cheeses.png"
]
   
blurred_images = [
    "./data/images/input/example/blurred-dairy_bottles.png",
    "./data/images/input/example/blurred-dairy_products.png",
    "./data/images/input/example/blurred-product_cheeses.png"
]

baseline_images_list = utils.processImages(baseline_images)
blurred_images_list = utils.processImages(blurred_images)

The Wallaroo SDK is capable of using numpy arrays in a pandas DataFrame for inference requests. Our demonstration will focus on using API calls for inference requests, so we will flatten the numpy array and use that value for our inference inputs. The following examples show using a baseline and blurred image for inference requests and the sample outputs.

# baseline image
df_test = pd.DataFrame({'tensor': wallaroo.utils.flatten_np_array_columns(baseline_images_list[0], 'tensor')})

df_test

result = pipeline.infer(df_test, 
                        dataset=['time', 'out.avg_confidence','out.avg_px_intensity','out.boxes','out.classes','out.confidences','check_failures','metadata'])
result

# example of an inference from a bad photo

df_test = pd.DataFrame({'tensor': wallaroo.utils.flatten_np_array_columns(blurred_images_list[0], 'tensor')})

result = pipeline.infer(df_test, dataset=['time', 'out.avg_confidence','out.avg_px_intensity','out.boxes','out.classes','out.confidences','check_failures','metadata'])
result

Inference Request via API

The following code performs the same inference request through the pipeline’s inference URL. This is used to demonstrate how the API inference result appears, which is used for the later examples.

headers = wl.auth.auth_header()

headers['Content-Type'] = 'application/json; format=pandas-records'

deploy_url = pipeline._deployment._url()

response = requests.post(
                    deploy_url, 
                    headers=headers, 
                    data=df_test.to_json(orient="records")
                )

display(pd.DataFrame(response.json()))

Store the Ops Pipeline Partition

Wallaroo pipeline logs include the metadata.partition field that indicates what instance of the pipeline performed the inference. This partition name updates each time a new pipeline version is created; modifying the pipeline steps and other actions changes the pipeline version.

• Reference
  • Wallaroo SDK Essentials Guide: Pipeline Log Management

ops_partition = result.loc[0, 'metadata.partition']
ops_partition

'engine-fc59fccbc-bs2sg'

API Inference Helper Functions

The following helper functions are set up to perform inferences through either the Wallaroo Ops pipeline, or an edge deployed version of the pipeline.

For this example, update the hostname testboy.local to the hostname of the deployed edge device.

import requests

def ops_pipeline_inference(df):
    df_flattened = pd.DataFrame({'tensor': wallaroo.utils.flatten_np_array_columns(df, 'tensor')})
    # api based inference request
    headers = wl.auth.auth_header()

    headers['Content-Type'] = 'application/json; format=pandas-records'

    deploy_url = pipeline._deployment._url()

    response = requests.post(
                        deploy_url, 
                        headers=headers, 
                        data=df_flattened.to_json(orient="records")
                    )

    display(pd.DataFrame(response.json()).loc[:, ['time', 'metadata']])

def edge_pipeline_inference(df):
    df_flattened = pd.DataFrame({'tensor': wallaroo.utils.flatten_np_array_columns(df, 'tensor')})
    # api based inference request
    # headers = wl.auth.auth_header()

    headers = {
        'Content-Type': 'application/json; format=pandas-records'
    }

    deploy_url = 'http://testboy.local:8080/pipelines/retail-inv-tracker-edge-obs'

    response = requests.post(
                        deploy_url, 
                        headers=headers, 
                        data=df_flattened.to_json(orient="records")
                    )

    display(pd.DataFrame(response.json()).loc[:, ['time', 'out', 'metadata']])

Edge Deployment

We can now deploy the pipeline to an edge device. This will require the following steps:

• Publish the pipeline: Publishes the pipeline to the OCI registry.
• Add Edge: Add the edge location to the pipeline publish.
• Deploy Edge: Deploy the edge device with the edge location settings.

pup = pipeline.publish()

Waiting for pipeline publish... It may take up to 600 sec.
Pipeline is Publishing......Published.

display(pup)

Add Edge

The edge location is added with the publish.add_edge(name) method. This returns the OCI registration information, and the EDGE_BUNDLE information. The EDGE_BUNDLE data is a base64 encoded set of parameters for the pipeline that the edge device is associated with.

Edge names must be unique. Update the edge name below if required.

edge_name = 'cv-observability-demo-sample'
edge_publish = pup.add_edge(edge_name)
display(edge_publish)

DevOps Deployment

The edge deployment is performed with docker run, docker compose, or helm installations. The following command generates the docker run command, with the following values provided by the DevOps Engineer:

• $REGISTRYURL
• $REGISTRYUSERNAME
• $REGISTRYPASSWORD

Before deploying, create the ./data directory that is used to store the authentication credentials.

# create docker run 

docker_command = f'''
docker run -p 8080:8080 \\
    -v ./data:/persist \\
    -e DEBUG=true \\
    -e OCI_REGISTRY=$REGISTRYURL \\
    -e EDGE_BUNDLE={edge_publish.docker_run_variables['EDGE_BUNDLE']} \\
    -e CONFIG_CPUS=6 \\
    -e OCI_USERNAME=$REGISTRYUSERNAME \\
    -e OCI_PASSWORD=$REGISTRYPASSWORD \\
    -e PIPELINE_URL={edge_publish.pipeline_url} \\
    {edge_publish.engine_url}
'''

print(docker_command)

docker run -p 8080:8080 \
    -v ./data:/persist \
    -e DEBUG=true \
    -e OCI_REGISTRY=$REGISTRYURL \
    -e EDGE_BUNDLE=ZXhwb3J0IEJVTkRMRV9WRVJTSU9OPTEKZXhwb3J0IEVER0VfTkFNRT1jdi1vYnNlcnZhYmlsaXR5LWRlbW8tc2FtcGxlCmV4cG9ydCBKT0lOX1RPS0VOPWQxN2E0NzJlLTYzNjQtNDcxMi05MWUyLWRhNzUzMjQ1MzNiYwpleHBvcnQgT1BTQ0VOVEVSX0hPU1Q9ZG9jLXRlc3QuZWRnZS53YWxsYXJvb2NvbW11bml0eS5uaW5qYQpleHBvcnQgUElQRUxJTkVfVVJMPWdoY3IuaW8vd2FsbGFyb29sYWJzL2RvYy1zYW1wbGVzL3BpcGVsaW5lcy9yZXRhaWwtaW52LXRyYWNrZXItZWRnZS1vYnM6YzJkZDJjN2QtNjE4ZC00OTdmLTg1MTUtM2FhMTQ2OWQ3OTg1CmV4cG9ydCBXT1JLU1BBQ0VfSUQ9OA== \
    -e CONFIG_CPUS=6 \
    -e OCI_USERNAME=$REGISTRYUSERNAME \
    -e OCI_PASSWORD=$REGISTRYPASSWORD \
    -e PIPELINE_URL=ghcr.io/wallaroolabs/doc-samples/pipelines/retail-inv-tracker-edge-obs:c2dd2c7d-618d-497f-8515-3aa1469d7985 \
    ghcr.io/wallaroolabs/doc-samples/engines/proxy/wallaroo/ghcr.io/wallaroolabs/standalone-mini:v2023.4.1-4351

Verify Logs

Before we perform inferences on the edge deployment, we’ll collect the pipeline logs and display the current partitions. These should only include the Wallaroo Ops pipeline.

logs = pipeline.logs(dataset=['time', 'metadata'])

ops_locations = [pd.unique(logs['metadata.partition']).tolist()][0]
display(ops_locations)
ops_location = ops_locations[0]

Warning: The inference log is above the allowable limit and the following columns may have been suppressed for various rows in the logs: ['in.tensor']. To review the dropped columns for an individual inference’s suppressed data, include dataset=["metadata"] in the log request.

['engine-fc59fccbc-bs2sg']

Drift Detection Example

The following uses our baseline and blurred data to create observability values. We start with creating a set of baseline images inference results through our deployed Ops pipeline, storing the start and end dates.

baseline_start = datetime.datetime.now(datetime.timezone.utc)

for i in range(10):
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)

time.sleep(10)

baseline_end = datetime.datetime.now(datetime.timezone.utc)

Build Assay with Baseline

We will use the baseline values to create our assay, specifying the start and end dates to use for the baseline values. From there we will run an interactive assay to view the current values against the baseline. Since they are just the baseline values, everything should be fine.

For our assay window, we will set the locations to both the Ops pipeline and the edge deployed pipeline. This gathers the pipeline log values from both partitions for the assay.

• References
  • Wallaroo SDK Essentials Guide: Assays Management

# create baseline from numpy

assay_name_from_dates = "average confidence drift detection example"
step_name = "resnet-with-intensity"
assay_builder_from_dates = wl.build_assay(assay_name_from_dates, 
                               pipeline, 
                               step_name, 
                               iopath="output avg_confidence 0", 
                               baseline_start=baseline_start,
                               baseline_end=baseline_end)

# assay from recent updates

assay_builder_from_dates = assay_builder_from_dates.add_run_until(baseline_end)

# View 1 minute intervals
# just ops
(assay_builder_from_dates
    .window_builder()
    .add_width(minutes=1)
    .add_interval(minutes=1)
    .add_start(baseline_start)
    .add_location_filter([ops_partition, edge_name])
)

<wallaroo.assay_config.WindowBuilder at 0x14850dd00>

assay_config = assay_builder_from_dates.build()
assay_results = assay_config.interactive_run()

print(f"Generated {len(assay_results)} analyses")
assay_results.chart_scores()

Generated 6 analyses

Set Observability Data

The next set of inferences will send all blurred image data to the edge deployed pipeline, and all good to the Ops pipeline. By the end, we will be able to demonstrate viewing the assay results to show detecting the blurred images causing more and more scores outside of the baseline.

Every so often we will rerun the interactive assay to show the updated results.

assay_window_start = datetime.datetime.now(datetime.timezone.utc)

for i in range(1):
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)
for i in range(9):
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)

# assay window from dates

assay_window_end = datetime.datetime.now(datetime.timezone.utc)

assay_builder_from_dates = assay_builder_from_dates.add_run_until(assay_window_end)

# View 1 minute intervals
# just combined
(assay_builder_from_dates
    .window_builder()
    .add_width(minutes=6)
    .add_interval(minutes=6)
    .add_start(assay_window_start)
    .add_location_filter([ops_partition, edge_name])
)

<wallaroo.assay_config.WindowBuilder at 0x14850dd00>

assay_config = assay_builder_from_dates.build()
assay_results = assay_config.interactive_run()

print(f"Generated {len(assay_results)} analyses")
assay_results.chart_scores()

Generated 1 analyses

for i in range(8):
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(2):
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

for i in range(7):
    print(i)
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(3):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

for i in range(6):
    print(i)
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(4):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

# assay window from dates

assay_window_end = datetime.datetime.now(datetime.timezone.utc)

assay_builder_from_dates = assay_builder_from_dates.add_run_until(assay_window_end)

# View 1 minute intervals
# just combined
(assay_builder_from_dates
    .window_builder()
    .add_width(minutes=7)
    .add_interval(minutes=7)
    .add_start(assay_window_start)
    .add_location_filter([ops_partition, edge_name])
)

<wallaroo.assay_config.WindowBuilder at 0x14850dd00>

assay_config = assay_builder_from_dates.build()
assay_results = assay_config.interactive_run()

print(f"Generated {len(assay_results)} analyses")
assay_results.chart_scores()

Generated 7 analyses

for i in range(5):
    print(i)
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(5):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

for i in range(4):
    print(i)
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(6):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

for i in range(3):
    print(i)
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(7):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

for i in range(2):
    print(i)
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(8):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

for i in range(1):
    print(i)
    for good_image in baseline_images_list:
        ops_pipeline_inference(good_image)
for i in range(9):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

for i in range(10):
    print(i)
    for bad_image in blurred_images_list:
        edge_pipeline_inference(bad_image)

# assay window from dates

assay_window_end = datetime.datetime.now(datetime.timezone.utc)

assay_builder_from_dates = assay_builder_from_dates.add_run_until(assay_window_end)

# just combined
(assay_builder_from_dates
    .window_builder()
    .add_width(minutes=6)
    .add_interval(minutes=6)
    .add_start(assay_window_start)
    .add_location_filter([ops_partition, edge_name])
)

<wallaroo.assay_config.WindowBuilder at 0x14850dd00>

assay_config = assay_builder_from_dates.build()
assay_results = assay_config.interactive_run()

print(f"Generated {len(assay_results)} analyses")
assay_results.chart_scores()

Generated 16 analyses

If we isolate to just the Ops center pipeline, we see a different result.

# assay window from dates

assay_window_end = datetime.datetime.now(datetime.timezone.utc)

assay_builder_from_dates = assay_builder_from_dates.add_run_until(assay_window_end)

# just combined
(assay_builder_from_dates
    .window_builder()
    .add_width(minutes=6)
    .add_interval(minutes=6)
    .add_start(assay_window_start)
    .add_location_filter([ops_partition])
)

<wallaroo.assay_config.WindowBuilder at 0x14850dd00>

assay_config = assay_builder_from_dates.build()
assay_results = assay_config.interactive_run()

print(f"Generated {len(assay_results)} analyses")
assay_results.chart_scores()

Generated 13 analyses

If we isolate to only the edge location, we see where the out of baseline scores are coming from.

# assay window from dates

assay_window_end = datetime.datetime.now(datetime.timezone.utc)

assay_builder_from_dates = assay_builder_from_dates.add_run_until(assay_window_end)

# just combined
(assay_builder_from_dates
    .window_builder()
    .add_width(minutes=6)
    .add_interval(minutes=6)
    .add_start(assay_window_start)
    .add_location_filter([edge_name])
)

<wallaroo.assay_config.WindowBuilder at 0x14850dd00>

assay_config = assay_builder_from_dates.build()
assay_results = assay_config.interactive_run()

print(f"Generated {len(assay_results)} analyses")
assay_results.chart_scores()

Generated 12 analyses

With the demonstration complete, we can shut down the edge deployed pipeline and undeploy the pipeline in the Ops center.

pipeline.undeploy()

2 - Wallaroo Model Observability: Anomaly Detection with CCFraud

The following tutorials are available from the Wallaroo Tutorials Repository.

Wallaroo Model Observability: Anomaly Detection with CCFraud

The following tutorial demonstrates the use case of detecting anomalies: inference input or output data that does not match typical validations.

Wallaroo provides validations to detect anomalous data from inference inputs and outputs. Validations are added to a Wallaroo pipeline with the wallaroo.pipeline.add_validations method.

Adding validations takes the format:

pipeline.add_validations(
    validation_name_01 = polars.col(in|out.{column_name}) EXPRESSION,
    validation_name_02 = polars.col(in|out.{column_name}) EXPRESSION
    ...{additional rules}
)

• validation_name: The user provided name of the validation. The names must match Python variable naming requirements.
  • IMPORTANT NOTE: Using the name count as a validation name returns an error. Any validation rules named count are dropped upon request and a warning returned.
• polars.col(in|out.{column_name}): Specifies the input or output for a specific field aka “column” in an inference result. Wallaroo inference requests are in the format in.{field_name} for inputs, and out.{field_name} for outputs.
  • More than one field can be selected, as long as they follow the rules of the polars 0.18 Expressions library.
• EXPRESSION: The expression to validate. When the expression returns True, that indicates an anomaly detected.

The polars library version 0.18.5 is used to create the validation rule. This is installed by default with the Wallaroo SDK. This provides a powerful range of comparisons to organizations tracking anomalous data from their ML models.

When validations are added to a pipeline, inference request outputs return the following fields:

When validation returns True, an anomaly is detected.

For example, adding the validation fraud to the following pipeline returns anomaly.count of 1 when the validation fraud returns True. The validation fraud returns True when the output field dense_1 at index 0 is greater than 0.9.

sample_pipeline = wallaroo.client.build_pipeline("sample-pipeline")
sample_pipeline.add_model_step(ccfraud_model)

# add the validation
sample_pipeline.add_validations(
    fraud=pl.col("out.dense_1").list.get(0) > 0.9,
    )

# deploy the pipeline
sample_pipeline.deploy()

# sample inference
display(sample_pipeline.infer_from_file("dev_high_fraud.json", data_format='pandas-records'))

Detecting Anomalies from Inference Request Results

When an inference request is submitted to a Wallaroo pipeline with validations, the following fields are output:

For example, adding the validation fraud to the following pipeline returns anomaly.count of 1 when the validation fraud returns True.

sample_pipeline = wallaroo.client.build_pipeline("sample-pipeline")
sample_pipeline.add_model_step(ccfraud_model)

# add the validation
sample_pipeline.add_validations(
    fraud=pl.col("out.dense_1").list.get(0) > 0.9,
    )

# deploy the pipeline
sample_pipeline.deploy()

# sample inference
display(sample_pipeline.infer_from_file("dev_high_fraud.json", data_format='pandas-records'))

Anomaly Detection Demonstration

The following demonstrates how to:

• Upload a ccfraud ML model trained to detect the likelihood of a transaction being fraudulent. This outputs the field dense_1 as an float where the closer to 1, the higher the likelihood of the transaction being fraudulent.
• Add the ccfraud model as a pipeline step.
• Add the validation fraud to detect when the output of dense_1 at index 0 when the values are greater than 0.9.
• Deploy the pipeline and performing sample inferences on it.
• Perform sample inferences to show when the fraud validation returns True and False.
• Perform sample inference with different datasets to show enable or disable certain fields from displaying in the inference results.

Prerequisites

• Wallaroo version 2023.4.1 and above.
• polars version 0.18.5. This is installed by default with the Wallaroo SDK.

Tutorial Steps

Load Libraries

The first step is to import the libraries used in this notebook.

import wallaroo
wallaroo.__version__

'2023.4.1+379cb6b8a'

Connect to the Wallaroo Instance through the User Interface

The next 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.

wl = wallaroo.Client()

Create a New Workspace

We’ll use the SDK below to create our workspace then assign as our current workspace. The current workspace is used by the Wallaroo SDK for where to upload models, create pipelines, etc. We’ll also set up variables for our models and pipelines down the road, so we have one spot to change names to whatever fits your organization’s standards best.

Before starting, verify that the workspace name is unique in your Wallaroo instance.

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

workspace_name = 'validation-ccfraud-demonstration-jch'
pipeline_name = 'ccfraud-validation-demo'
model_name = 'ccfraud'
model_file_name = './models/ccfraud.onnx'

workspace = get_workspace(workspace_name, wl)
wl.set_current_workspace(workspace)

{'name': 'validation-ccfraud-demonstration-jch', 'id': 19, 'archived': False, 'created_by': 'c97d480f-6064-4537-b18e-40fb1864b4cd', 'created_at': '2024-02-08T16:57:29.044902+00:00', 'models': [{'name': 'ccfraud', 'versions': 2, 'owner_id': '""', 'last_update_time': datetime.datetime(2024, 2, 8, 17, 12, 40, 341069, tzinfo=tzutc()), 'created_at': datetime.datetime(2024, 2, 8, 16, 57, 31, 612826, tzinfo=tzutc())}], 'pipelines': [{'name': 'ccfraud-validation', 'create_time': datetime.datetime(2024, 2, 8, 16, 57, 32, 340043, tzinfo=tzutc()), 'definition': '[]'}]}

Upload the Model

Upload the model to the Wallaroo workspace with the wallaroo.client.upload_model method. Our ccfraud ML model is a Wallaroo Default Runtime of type ONNX, so all we need is the model name, the model file path, and the framework type of wallaroo.framework.Framework.ONNX.

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

Build the Pipeline

Pipelines are build with the wallaroo.client.build_pipeline method, which takes the pipeline name. This will create the pipeline in our default workspace. Note that if there are any existing pipelines with the same name in this workspace, this method will retrieve that pipeline for this SDK session.

Once the pipeline is created, we add the ccfraud model as our pipeline step.

sample_pipeline = wl.build_pipeline(pipeline_name)
sample_pipeline = sample_pipeline.add_model_step(ccfraud_model)

Add Validation

Now we add our validation to our new pipeline. We will give it the following configuration.

• Validation Name: fraud
• Validation Field: out.dense_1
• Validation Field Index: 0
• Validation Expression: Values greater than 0.9.

The polars library is required for creating the validation. We will import the polars library, then add our validation to the pipeline.

• IMPORTANT NOTE: Validation names must be unique per pipeline. If a validation of the same name is added, both are included in the pipeline validations, but only most recent validation with the same name is displayed with the inference results. Anomalies detected by multiple validations of the same name are added to the anomaly.count inference result field.

import polars as pl

sample_pipeline = sample_pipeline.add_validations(
    fraud=pl.col("out.dense_1").list.get(0) > 0.9
)

Display Pipeline And Validation Steps

The method wallaroo.pipeline.steps() shows the current pipeline steps. The added validations are in the Check field. This is used for demonstration purposes to show the added validation to the pipeline.

sample_pipeline.steps()

[{'ModelInference': {'models': [{'name': 'ccfraud', 'version': 'f1f2ab86-a41d-4601-b14a-b594e3d86c6e', 'sha': 'bc85ce596945f876256f41515c7501c399fd97ebcb9ab3dd41bf03f8937b4507'}]}},
 {'Check': {'tree': ['{"Alias":[{"BinaryExpr":{"left":{"Function":{"input":[{"Column":"out.dense_1"},{"Literal":{"Int32":0}}],"function":{"ListExpr":"Get"},"options":{"collect_groups":"ApplyFlat","fmt_str":"","input_wildcard_expansion":false,"auto_explode":true,"cast_to_supertypes":false,"allow_rename":false,"pass_name_to_apply":false,"changes_length":false,"check_lengths":true,"allow_group_aware":true}}},"op":"Gt","right":{"Literal":{"Float64":0.9}}}},"fraud"]}']}}]

Deploy Pipeline

With the pipeline steps set and the validations created, we deploy the pipeline. Because of it’s size, we will only allocate 0.1 cpu from the cluster for the pipeline’s use.

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.1)\
    .build()

sample_pipeline.deploy(deployment_config=deploy_config)

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

Sample Inferences

Two sample inferences are performed with the method wallaroo.pipeline.infer_from_file that takes either a pandas Record JSON file or an Apache Arrow table as the input.

For our demonstration, we will use the following pandas Record JSON files with the following sample data:

• ./data/dev_smoke_test.pandas.json: A sample inference that generates a low (lower than 0.01) likelihood of fraud.
• ./data/dev_high_fraud.pandas.json: A sample inference that generates a high (higher than 0.90) likelihood of fraud.

The inference request returns a pandas DataFrame.

Each of the inference outputs will include the following fields:

sample_pipeline.infer_from_file("./data/dev_smoke_test.pandas.json")

sample_pipeline.infer_from_file("./data/dev_high_fraud.json")

Other Validation Examples

The following are additional examples of validations.

Multiple Validations

The following uses multiple validations to check for anomalies. We still use fraud which detects outputs that are greater than 0.9. The second validation too_low triggers an anomaly when the out.dense_1 is under 0.05.

After the validations are added, the pipeline is redeployed to “set” them.

sample_pipeline = sample_pipeline.add_validations(
    too_low=pl.col("out.dense_1").list.get(0) < 0.001
)

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.1)\
    .build()
sample_pipeline.undeploy()
sample_pipeline.deploy(deployment_config=deploy_config)

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

sample_pipeline.infer_from_file("./data/dev_smoke_test.pandas.json")

sample_pipeline.infer_from_file("./data/dev_high_fraud.json")

Compound Validations

The following combines multiple field checks into a single validation. For this, we will check for values of out.dense_1 that are between 0.05 and 0.9.

Each expression is separated by (). For example:

• Expression 1: pl.col("out.dense_1").list.get(0) < 0.9
• Expression 2: pl.col("out.dense_1").list.get(0) > 0.001
• Compound Expression: (pl.col("out.dense_1").list.get(0) < 0.9) & (pl.col("out.dense_1").list.get(0) > 0.001)

sample_pipeline = sample_pipeline.add_validations(
    in_between_2=(pl.col("out.dense_1").list.get(0) < 0.9) & (pl.col("out.dense_1").list.get(0) > 0.001)
)

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.1)\
    .build()
sample_pipeline.undeploy()
sample_pipeline.deploy(deployment_config=deploy_config)

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

results = sample_pipeline.infer_from_file("./data/cc_data_1k.df.json")

results.loc[results['anomaly.in_between_2'] == True] 

179 rows × 7 columns

Specify Dataset

Wallaroo inference requests allow datasets to be excluded or included with the dataset_exclude and dataset parameters.

For our example, we will exclude the anomaly dataset, but include the datasets 'time', 'in', 'out', 'anomaly.count'. Note that while we exclude anomaly, we override that with by setting the anomaly field 'anomaly.count' in our dataset parameter.

sample_pipeline.infer_from_file("./data/dev_high_fraud.json", 
                                dataset_exclude=['anomaly'], 
                                dataset=['time', 'in', 'out', 'anomaly.count']
                                )

Undeploy the Pipeline

With the demonstration complete, we undeploy the pipeline and return the resources back to the cluster.

sample_pipeline.undeploy()

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

3 - Wallaroo Model Observability: Anomaly Detection with House Price Prediction

The following tutorials are available from the Wallaroo Tutorials Repository.

Wallaroo Model Observability: Anomaly Detection with House Price Prediction

The following tutorial demonstrates the use case of detecting anomalies: inference input or output data that does not match typical validations.

Wallaroo provides validations to detect anomalous data from inference inputs and outputs. Validations are added to a Wallaroo pipeline with the wallaroo.pipeline.add_validations method.

Adding validations takes the format:

pipeline.add_validations(
    validation_name_01 = polars.col(in|out.{column_name}) EXPRESSION,
    validation_name_02 = polars.col(in|out.{column_name}) EXPRESSION
    ...{additional rules}
)

• validation_name: The user provided name of the validation. The names must match Python variable naming requirements.
  • IMPORTANT NOTE: Using the name count as a validation name returns an error. Any validation rules named count are dropped upon request and a warning returned.
• polars.col(in|out.{column_name}): Specifies the input or output for a specific field aka “column” in an inference result. Wallaroo inference requests are in the format in.{field_name} for inputs, and out.{field_name} for outputs.
  • More than one field can be selected, as long as they follow the rules of the polars 0.18 Expressions library.
• EXPRESSION: The expression to validate. When the expression returns True, that indicates an anomaly detected.

The polars library version 0.18.5 is used to create the validation rule. This is installed by default with the Wallaroo SDK. This provides a powerful range of comparisons to organizations tracking anomalous data from their ML models.

When validations are added to a pipeline, inference request outputs return the following fields:

When validation returns True, an anomaly is detected.

For example, adding the validation fraud to the following pipeline returns anomaly.count of 1 when the validation fraud returns True. The validation fraud returns True when the output field dense_1 at index 0 is greater than 0.9.

sample_pipeline = wallaroo.client.build_pipeline("sample-pipeline")
sample_pipeline.add_model_step(model)

# add the validation
sample_pipeline.add_validations(
    fraud=pl.col("out.dense_1").list.get(0) > 0.9,
    )

# deploy the pipeline
sample_pipeline.deploy()

# sample inference
display(sample_pipeline.infer_from_file("dev_high_fraud.json", data_format='pandas-records'))

Detecting Anomalies from Inference Request Results

When an inference request is submitted to a Wallaroo pipeline with validations, the following fields are output:

For example, adding the validation fraud to the following pipeline returns anomaly.count of 1 when the validation fraud returns True.

sample_pipeline = wallaroo.client.build_pipeline("sample-pipeline")
sample_pipeline.add_model_step(model)

# add the validation
sample_pipeline.add_validations(
    fraud=pl.col("out.dense_1").list.get(0) > 0.9,
    )

# deploy the pipeline
sample_pipeline.deploy()

# sample inference
display(sample_pipeline.infer_from_file("dev_high_fraud.json", data_format='pandas-records'))

Anomaly Detection Demonstration

The following demonstrates how to:

• Upload a house price ML model trained to predict house prices based on a set of inputs. This outputs the field variable as an float which is the predicted house price.
• Add the house price model as a pipeline step.
• Add the validation too_high to detect when a house price exceeds a certain value.
• Deploy the pipeline and performing sample inferences on it.
• Perform sample inferences to show when the too_high validation returns True and False.
• Perform sample inference with different datasets to show enable or disable certain fields from displaying in the inference results.

Prerequisites

• Wallaroo version 2023.4.1 and above.
• polars version 0.18.5. This is installed by default with the Wallaroo SDK.

Tutorial Steps

Load Libraries

The first step is to import the libraries used in this notebook.

import wallaroo
wallaroo.__version__

'2023.4.1+379cb6b8a'

Connect to the Wallaroo Instance through the User Interface

The next 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.

wl = wallaroo.Client()

Create a New Workspace

We’ll use the SDK below to create our workspace then assign as our current workspace. The current workspace is used by the Wallaroo SDK for where to upload models, create pipelines, etc. We’ll also set up variables for our models and pipelines down the road, so we have one spot to change names to whatever fits your organization’s standards best.

Before starting, verify that the workspace name is unique in your Wallaroo instance.

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

workspace_name = 'validation-house-price-demonstration'
pipeline_name = 'validation-demo'
model_name = 'anomaly-housing-model'
model_file_name = './models/rf_model.onnx'

workspace = get_workspace(workspace_name, wl)
wl.set_current_workspace(workspace)

{'name': 'validation-house-price-demonstration', 'id': 25, 'archived': False, 'created_by': 'c97d480f-6064-4537-b18e-40fb1864b4cd', 'created_at': '2024-02-08T21:52:50.354176+00:00', 'models': [{'name': 'anomaly-housing-model', 'versions': 1, 'owner_id': '""', 'last_update_time': datetime.datetime(2024, 2, 8, 21, 52, 51, 671284, tzinfo=tzutc()), 'created_at': datetime.datetime(2024, 2, 8, 21, 52, 51, 671284, tzinfo=tzutc())}], 'pipelines': [{'name': 'validation-demo', 'create_time': datetime.datetime(2024, 2, 8, 21, 52, 52, 879885, tzinfo=tzutc()), 'definition': '[]'}]}

Upload the Model

Upload the model to the Wallaroo workspace with the wallaroo.client.upload_model method. Our house price ML model is a Wallaroo Default Runtime of type ONNX, so all we need is the model name, the model file path, and the framework type of wallaroo.framework.Framework.ONNX.

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

Build the Pipeline

Pipelines are build with the wallaroo.client.build_pipeline method, which takes the pipeline name. This will create the pipeline in our default workspace. Note that if there are any existing pipelines with the same name in this workspace, this method will retrieve that pipeline for this SDK session.

Once the pipeline is created, we add the ccfraud model as our pipeline step.

sample_pipeline = wl.build_pipeline(pipeline_name)
sample_pipeline.clear()
sample_pipeline = sample_pipeline.add_model_step(model)

import onnx

model = onnx.load(model_file_name)
output =[node.name for node in model.graph.output]

input_all = [node.name for node in model.graph.input]
input_initializer =  [node.name for node in model.graph.initializer]
net_feed_input = list(set(input_all)  - set(input_initializer))

print('Inputs: ', net_feed_input)
print('Outputs: ', output)

Inputs:  ['float_input']
Outputs:  ['variable']

Add Validation

Now we add our validation to our new pipeline. We will give it the following configuration.

• Validation Name: too_high
• Validation Field: out.variable
• Validation Field Index: 0
• Validation Expression: Values greater than 1000000.0.

The polars library is required for creating the validation. We will import the polars library, then add our validation to the pipeline.

• IMPORTANT NOTE: Validation names must be unique per pipeline. If a validation of the same name is added, both are included in the pipeline validations, but only most recent validation with the same name is displayed with the inference results. Anomalies detected by multiple validations of the same name are added to the anomaly.count inference result field.

import polars as pl

sample_pipeline = sample_pipeline.add_validations(
    too_high=pl.col("out.variable").list.get(0) > 1000000.0
)

Display Pipeline And Validation Steps

The method wallaroo.pipeline.steps() shows the current pipeline steps. The added validations are in the Check field. This is used for demonstration purposes to show the added validation to the pipeline.

sample_pipeline.steps()

[{'ModelInference': {'models': [{'name': 'anomaly-housing-model', 'version': '9a76a2cf-9ea3-4978-8fd5-005d0280e661', 'sha': 'e22a0831aafd9917f3cc87a15ed267797f80e2afa12ad7d8810ca58f173b8cc6'}]}},
 {'Check': {'tree': ['{"Alias":[{"BinaryExpr":{"left":{"Function":{"input":[{"Column":"out.variable"},{"Literal":{"Int32":0}}],"function":{"ListExpr":"Get"},"options":{"collect_groups":"ApplyFlat","fmt_str":"","input_wildcard_expansion":false,"auto_explode":true,"cast_to_supertypes":false,"allow_rename":false,"pass_name_to_apply":false,"changes_length":false,"check_lengths":true,"allow_group_aware":true}}},"op":"Gt","right":{"Literal":{"Float64":1000000.0}}}},"too_high"]}']}}]

Deploy Pipeline

With the pipeline steps set and the validations created, we deploy the pipeline. Because of it’s size, we will only allocate 0.1 cpu from the cluster for the pipeline’s use.

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.25)\
    .build()

sample_pipeline.deploy(deployment_config=deploy_config)

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

Sample Inferences

Two sample inferences are performed with the method wallaroo.pipeline.infer_from_file that takes either a pandas Record JSON file or an Apache Arrow table as the input.

For our demonstration, we will use the following pandas Record JSON file with the following sample data:

• ./data/houseprice_5000_data.json: A sample sets of 5000 houses to generates a range of predicted values.

The inference request returns a pandas DataFrame.

Each of the inference outputs will include the following fields:

results = sample_pipeline.infer_from_file('./data/test-1000.df.json')
# first 20 results
display(results.head(20))

# only results that trigger the anomaly too_high
results.loc[results['anomaly.too_high'] == True]

Other Validation Examples

The following are additional examples of validations.

Multiple Validations

The following uses multiple validations to check for anomalies. We still use fraud which detects outputs that are greater than 1000000.0. The second validation too_low triggers an anomaly when the out.variable is under 250000.0.

After the validations are added, the pipeline is redeployed to “set” them.

sample_pipeline = sample_pipeline.add_validations(
    too_low=pl.col("out.variable").list.get(0) < 250000.0
)

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.1)\
    .build()
sample_pipeline.undeploy()
sample_pipeline.deploy(deployment_config=deploy_config)

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

results = sample_pipeline.infer_from_file('./data/test-1000.df.json')
# first 20 results
display(results.head(20))

# only results that trigger the anomaly too_high
results.loc[results['anomaly.too_high'] == True]

# only results that trigger the anomaly too_low
results.loc[results['anomaly.too_low'] == True]

62 rows × 6 columns

Compound Validations

The following combines multiple field checks into a single validation. For this, we will check for values of out.variable that are between 500000 and 1000000.

Each expression is separated by (). For example:

• Expression 1: pl.col("out.variable").list.get(0) < 1000000.0
• Expression 2: pl.col("out.variable").list.get(0) > 500000.0
• Compound Expression: (pl.col("out.variable").list.get(0) < 1000000.0) & (pl.col("out.variable").list.get(0) > 500000.0)

sample_pipeline = sample_pipeline.add_validations(
    in_between=(pl.col("out.variable").list.get(0) < 1000000.0) & (pl.col("out.variable").list.get(0) > 500000.0)
)

deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.1)\
    .build()
sample_pipeline.undeploy()
sample_pipeline.deploy(deployment_config=deploy_config)

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

results = sample_pipeline.infer_from_file('./data/test-1000.df.json')

results.loc[results['anomaly.in_between'] == True] 

395 rows × 7 columns

Specify Dataset

Wallaroo inference requests allow datasets to be excluded or included with the dataset_exclude and dataset parameters.

For our example, we will exclude the anomaly dataset, but include the datasets 'time', 'in', 'out', 'anomaly.count'. Note that while we exclude anomaly, we override that with by setting the anomaly field 'anomaly.count' in our dataset parameter.

sample_pipeline.infer_from_file('./data/test-1000.df.json', 
                                dataset_exclude=['anomaly'], 
                                dataset=['time', 'in', 'out', 'anomaly.count']
                                )

1000 rows × 4 columns

Undeploy the Pipeline

With the demonstration complete, we undeploy the pipeline and return the resources back to the cluster.

sample_pipeline.undeploy()

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

4 - Wallaroo Edge Observability with Classification Financial Models

The following tutorial is available on the Wallaroo Github Repository.

Classification Financial Services Edge Deployment Demonstration

This notebook will walk through building Wallaroo pipeline with a a Classification model deployed to detect the likelihood of credit card fraud, then publishing that pipeline to an Open Container Initiative (OCI) Registry where it can be deployed in other Docker and Kubernetes environments.

This demonstration will focus on deployment to the edge. For further examples of using Wallaroo with this computer vision models, see Wallaroo 101.

This demonstration performs the following:

• In Wallaroo Ops:
  • Setting up a workspace, pipeline, and model for deriving the price of a house based on inputs.
  • Creating an assay from a sample of inferences.
  • Display the inference result and upload the assay to the Wallaroo instance where it can be referenced later.
• In a remote aka edge location:
  • Deploying the Wallaroo pipeline as a Wallaroo Inference Server deployed on an edge device with observability features.
• In Wallaroo Ops:
  • Observe the Wallaroo Ops and remote Wallaroo Inference Server inference results as part of the pipeline logs.

Prerequisites

• A deployed Wallaroo Ops instance.
• A location with Docker or Kubernetes with helm for Wallaroo Inference server deployments.
• The following Python libraries installed:
  • wallaroo: The Wallaroo SDK. Included with the Wallaroo JupyterHub service by default.
  • pandas: Pandas, mainly used for Pandas DataFrame

References

• Wallaroo SDK Essentials Guide: Pipeline Edge Publication

Data Scientist Pipeline Publish Steps

Load Libraries

The first step is to import the libraries used in this notebook.

import wallaroo
from wallaroo.object import EntityNotFoundError

import pyarrow as pa
import pandas as pd
import requests

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

Connect to the Wallaroo Instance through the User Interface

The next 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.

wl = wallaroo.Client()

Create a New Workspace

We’ll use the SDK below to create our workspace , assign as our current workspace, then display all of the workspaces we have at the moment. We’ll also set up variables for our models and pipelines down the road, so we have one spot to change names to whatever fits your organization’s standards best.

To allow this tutorial to be run by multiple users in the same Wallaroo instance, a random 4 character prefix will be added to the workspace, pipeline, and model. Feel free to set suffix='' if this is not required.

import string
import random

# make a random 4 character suffix to verify uniqueness in tutorials
random_suffix= ''.join(random.choice(string.ascii_lowercase) for i in range(4))

suffix='jch'

workspace_name = f'edge-observability-demo{suffix}'
pipeline_name = 'edge-observability-pipeline'
xgboost_model_name = 'ccfraud-xgboost'
xgboost_model_file_name = './models/xgboost_ccfraud.onnx'

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

workspace = get_workspace(workspace_name)

wl.set_current_workspace(workspace)

{'name': 'edge-observability-demojch', 'id': 14, 'archived': False, 'created_by': 'b3deff28-04d0-41b8-a04f-b5cf610d6ce9', 'created_at': '2023-10-31T20:12:45.456363+00:00', 'models': [], 'pipelines': []}

Upload the Model

When a model is uploaded to a Wallaroo cluster, it is optimized and packaged to make it ready to run as part of a pipeline. In many times, the Wallaroo Server can natively run a model without any Python overhead. In other cases, such as a Python script, a custom Python environment will be automatically generated. This is comparable to the process of “containerizing” a model by adding a small HTTP server and other wrapping around it.

Our pretrained model is in ONNX format, which is specified in the framework parameter.

xgboost_edge_demo_model = wl.upload_model(
    xgboost_model_name,
    xgboost_model_file_name,
    framework=wallaroo.framework.Framework.ONNX,
).configure(tensor_fields=["tensor"])

Reserve Pipeline Resources

Before deploying an inference engine we need to tell wallaroo what resources it will need.
To do this we will use the wallaroo DeploymentConfigBuilder() and fill in the options listed below to determine what the properties of our inference engine will be.

We will be testing this deployment for an edge scenario, so the resource specifications are kept small – what’s the minimum needed to meet the expected load on the planned hardware.

• cpus - 1 => allow the engine to use 4 CPU cores when running the neural net
• memory - 900Mi => each inference engine will have 2 GB of memory, which is plenty for processing a single image at a time.
• arch - we will specify the X86 architecture.

deploy_config = (wallaroo
                 .DeploymentConfigBuilder()
                 .replica_count(1)
                 .cpus(1)
                 .memory("900Mi")
                 .build()
                 )

Simulated Edge Deployment

We will now deploy our pipeline into the current Kubernetes environment using the specified resource constraints. This is a “simulated edge” deploy in that we try to mimic the edge hardware as closely as possible.

pipeline = get_pipeline(pipeline_name)
display(pipeline)

# clear the pipeline if previously run
pipeline.clear()
pipeline.add_model_step(xgboost_edge_demo_model)

pipeline.deploy(deployment_config = deploy_config)

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

Run Single Image Inference

A single image, encoded using the Apache Arrow format, is sent to the deployed pipeline. Arrow is used here because, as a binary protocol, there is far lower network and compute overhead than using JSON. The Wallaroo Server engine accepts both JSON, pandas DataFrame, and Apache Arrow formats.

The sample DataFrames and arrow tables are in the ./data directory. We’ll use the Apache Arrow table cc_data_10k.arrow.

Once complete, we’ll display how long the inference request took.

import datetime
import time

deploy_url = pipeline._deployment._url()

headers = wl.auth.auth_header()

headers['Content-Type']='application/vnd.apache.arrow.file'
# headers['Content-Type']='application/json; format=pandas-records'
headers['Accept']='application/json; format=pandas-records'

dataFile = './data/cc_data_1k.arrow'
local_inference_start = datetime.datetime.now()

!curl -X POST {deploy_url} \
     -H "Authorization:{headers['Authorization']}" \
     -H "Content-Type:{headers['Content-Type']}" \
     -H "Accept:{headers['Accept']}" \
     --data-binary @{dataFile} > curl_response_xgboost.df.json

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100  799k  100  685k  100  114k  39.3M  6733k --:--:-- --:--:-- --:--:-- 45.9M

We will import the inference output, and isolate the metadata partition to store where the inference results are stored in the pipeline logs.

# display the first 20 results

df_results = pd.read_json('./curl_response_xgboost.df.json', orient="records")
# get just the partition
df_results['partition'] = df_results['metadata'].map(lambda x: x['partition'])
# display(df_results.head(20))
display(df_results.head(20).loc[:, ['time', 'out', 'partition']])

Undeploy Pipeline

With our testing complete, we will undeploy the pipeline and return the resources back to the cluster.

pipeline.undeploy()

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

Publish the Pipeline for Edge Deployment

It worked! For a demo, we’ll take working once as “tested”. So now that we’ve tested our pipeline, we are ready to publish it for edge deployment.

Publishing it means assembling all of the configuration files and model assets and pushing them to an Open Container Initiative (OCI) repository set in the Wallaroo instance as the Edge Registry service. DevOps engineers then retrieve that image and deploy it through Docker, Kubernetes, or similar deployments.

See Edge Deployment Registry Guide for details on adding an OCI Registry Service to Wallaroo as the Edge Deployment Registry.

This is done through the SDK command wallaroo.pipeline.publish(deployment_config) which has the following parameters and returns.

Publish a Pipeline Parameters

The publish method takes the following parameters. The containerized pipeline will be pushed to the Edge registry service with the model, pipeline configurations, and other artifacts needed to deploy the pipeline.

Publish a Pipeline Returns

Publish Example

We will now publish the pipeline to our Edge Deployment Registry with the pipeline.publish(deployment_config) command. deployment_config is an optional field that specifies the pipeline deployment. This can be overridden by the DevOps engineer during deployment.

pub=pipeline.publish(deploy_config)
display(pub)

Waiting for pipeline publish... It may take up to 600 sec.
Pipeline is Publishing....Published.

List Published Pipelines

The method wallaroo.client.list_pipelines() shows a list of all pipelines in the Wallaroo instance, and includes the published field that indicates whether the pipeline was published to the registry (True), or has not yet been published (False).

wl.list_pipelines()

List Publishes from a Pipeline

All publishes created from a pipeline are displayed with the wallaroo.pipeline.publishes method. The pipeline_version_id is used to know what version of the pipeline was used in that specific publish. This allows for pipelines to be updated over time, and newer versions to be sent and tracked to the Edge Deployment Registry service.

List Publishes Parameters

N/A

List Publishes Returns

A List of the following fields:

pipeline.publishes()

DevOps - Pipeline Edge Deployment

Once a pipeline is deployed to the Edge Registry service, it can be deployed in environments such as Docker, Kubernetes, or similar container running services by a DevOps engineer as a Wallaroo Server.

The following guides will demonstrate publishing a Wallaroo Pipeline as a Wallaroo Server.

Add Edge Location

Wallaroo Servers can optionally connect to the Wallaroo Ops instance and transmit their inference results. These are added to the pipeline logs for the published pipeline the Wallaroo Server is associated with.

Wallaroo Servers are added with the wallaroo.pipeline_publish.add_edge(name: string) method. The name is the unique primary key for each edge added to the pipeline publish and must be unique.

This returns a Publish Edge with the following fields:

Two edge publishes will be created so we can demonstrate removing an edge shortly.

edge_01_name = f'edge-ccfraud-observability{random_suffix}'
edge01 = pub.add_edge(edge_01_name)
display(edge01)

edge_02_name = f'edge-ccfraud-observability-02{random_suffix}'
edge02 = pub.add_edge(edge_02_name)
display(edge02)

pipeline.list_edges()

Remove Edge Location

Wallaroo Servers are removed with the wallaroo.pipeline_publish.remove_edge(name: string) method.

This returns a Publish Edge with the following fields:

Two edge publishes will be created so we can demonstrate removing an edge shortly.

sample = pub.remove_edge(edge_02_name)
display(sample)

None

pipeline.list_edges()

Docker Deployment

First, the DevOps engineer must authenticate to the same OCI Registry service used for the Wallaroo Edge Deployment registry.

For more details, check with the documentation on your artifact service. The following are provided for the three major cloud services:

• Set up authentication for Docker
• Authenticate with an Azure container registry
• Authenticating Amazon ECR Repositories for Docker CLI with Credential Helper

For the deployment, the engine URL is specified with the following environmental variables:

• DEBUG (true|false): Whether to include debug output.
• OCI_REGISTRY: The URL of the registry service.
• CONFIG_CPUS: The number of CPUs to use.
• OCI_USERNAME: The edge registry username.
• OCI_PASSWORD: The edge registry password or token.
• PIPELINE_URL: The published pipeline URL.

Docker Deployment Example

Using our sample environment, here’s sample deployment using Docker with a computer vision ML model, the same used in the Wallaroo Use Case Tutorials Computer Vision: Retail tutorials.

docker run -p 8080:8080 \
    -e DEBUG=true \
    -e OCI_REGISTRY={your registry server} \
    -e EDGE_BUNDLE={Your edge bundle['EDGE_BUNDLE']}
    -e CONFIG_CPUS=1 \
    -e OCI_USERNAME=oauth2accesstoken \
    -e OCI_PASSWORD={registry token here} \
    -e PIPELINE_URL={your registry server}{Your Wallaroo Server pipeline} \
    {your registry server}/{Your Wallaroo Server engine}

Docker Compose Deployment

For users who prefer to use docker compose, the following sample compose.yaml file is used to launch the Wallaroo Edge pipeline. This is the same used in the Wallaroo Use Case Tutorials Computer Vision: Retail tutorials.

services:
  engine:
    image: {Your Engine URL}
    ports:
      - 8080:8080
    environment:
      EDGE_BUNDLE: {Your Edge Bundle['EDGE_BUNDLE']}
      PIPELINE_URL: {Your Pipeline URL}
      OCI_REGISTRY: {Your Edge Registry URL}
      OCI_USERNAME:  {Your Registry Username}
      OCI_PASSWORD: {Your Token or Password}
      CONFIG_CPUS: 1

For example:

services:
  engine:
    image: sample-registry.com/engine:v2023.3.0-main-3707
    ports:
      - 8080:8080
    environment:
      PIPELINE_URL: sample-registry.com/pipelines/edge-cv-retail:bf70eaf7-8c11-4b46-b751-916a43b1a555
      OCI_REGISTRY: sample-registry.com
      OCI_USERNAME:  _json_key_base64
      OCI_PASSWORD: abc123
      CONFIG_CPUS: 1

Docker Compose Deployment Example

The deployment and undeployment is then just a simple docker compose up and docker compose down. The following shows an example of deploying the Wallaroo edge pipeline using docker compose.

docker compose up
[+] Running 1/1
 ✔ Container cv_data-engine-1  Recreated                                                                                                                                                                 0.5s
Attaching to cv_data-engine-1
cv_data-engine-1  | Wallaroo Engine - Standalone mode
cv_data-engine-1  | Login Succeeded
cv_data-engine-1  | Fetching manifest and config for pipeline: sample-registry.com/pipelines/edge-cv-retail:bf70eaf7-8c11-4b46-b751-916a43b1a555
cv_data-engine-1  | Fetching model layers
cv_data-engine-1  | digest: sha256:c6c8869645962e7711132a7e17aced2ac0f60dcdc2c7faa79b2de73847a87984
cv_data-engine-1  |   filename: c6c8869645962e7711132a7e17aced2ac0f60dcdc2c7faa79b2de73847a87984
cv_data-engine-1  |   name: resnet-50
cv_data-engine-1  |   type: model
cv_data-engine-1  |   runtime: onnx
cv_data-engine-1  |   version: 693e19b5-0dc7-4afb-9922-e3f7feefe66d
cv_data-engine-1  |
cv_data-engine-1  | Fetched
cv_data-engine-1  | Starting engine
cv_data-engine-1  | Looking for preexisting `yaml` files in //modelconfigs
cv_data-engine-1  | Looking for preexisting `yaml` files in //pipelines

Helm Deployment

Published pipelines can be deployed through the use of helm charts.

Helm deployments take up to two steps - the first step is in retrieving the required values.yaml and making updates to override.

1. Pull the helm charts from the published pipeline. The two fields are the Helm Chart URL and the Helm Chart version to specify the OCI . This typically takes the format of:

helm pull oci://{published.helm_chart_url} --version {published.helm_chart_version}

1. Extract the tgz file and copy the values.yaml and copy the values used to edit engine allocations, etc. The following are required for the deployment to run:

ociRegistry:
  registry: {your registry service}
  username:  {registry username here}
  password: {registry token here}

Store this into another file, suc as local-values.yaml.

1. Create the namespace to deploy the pipeline to. For example, the namespace wallaroo-edge-pipeline would be:

kubectl create -n wallaroo-edge-pipeline

1. Deploy the helm installation with helm install through one of the following options:

   1. Specify the tgz file that was downloaded and the local values file. For example:

      helm install --namespace {namespace} --values {local values file} {helm install name} {tgz path}
      

   2. Specify the expended directory from the downloaded tgz file.

      helm install --namespace {namespace} --values {local values file} {helm install name} {helm directory path}
      

   3. Specify the Helm Pipeline Helm Chart and the Pipeline Helm Version.

      helm install --namespace {namespace} --values {local values file} {helm install name} oci://{published.helm_chart_url} --version {published.helm_chart_version}
      

2. Once deployed, the DevOps engineer will have to forward the appropriate ports to the svc/engine-svc service in the specific pipeline. For example, using kubectl port-forward to the namespace ccfraud that would be:

   kubectl port-forward svc/engine-svc -n ccfraud01 8080 --address 0.0.0.0`