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

Pipeline Log Tutorial

This tutorial will demonstrate how to:

1. Select or create a workspace, pipeline and upload the control model, then additional models for A/B Testing and Shadow Deploy.
2. Add a pipeline step with the champion model, then deploy the pipeline and perform sample inferences.
3. Display the various log types for a standard deployed pipeline.
4. Swap out the pipeline step with the champion model with a shadow deploy step that compares the champion model against two competitors.
5. Perform sample inferences with a shadow deployed step, then display the log files for a shadow deployed pipeline.
6. Swap out the shadow deployed pipeline step with an A/B pipeline step.
7. Perform sample inferences with a A/B pipeline step, then display the log files for an A/B pipeline step.
8. Undeploy the pipeline.

This tutorial provides the following:

• Models:
  • models/rf_model.onnx: The champion model that has been used in this environment for some time.
  • models/xgb_model.onnx and models/gbr_model.onnx: Rival models that will be tested against the champion.
• 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.

Prerequisites

• A deployed Wallaroo instance
• The following Python libraries installed:
  • 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

Import libraries

The first step is to import the libraries needed for this notebook.

import wallaroo
from wallaroo.object import EntityNotFoundError

import pyarrow as pa

from IPython.display import display

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

import datetime

import os

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

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_name = 'logworkspace'
main_pipeline_name = 'logpipeline-test'
model_name_control = 'logcontrol'
model_file_name_control = './models/rf_model.onnx'

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

wl.set_current_workspace(workspace)

{'name': 'logworkspace', 'id': 10, 'archived': False, 'created_by': '65124b18-8382-49af-b3c8-ada3b9df3330', 'created_at': '2024-04-16T14:47:15.121824+00:00', 'models': [], 'pipelines': []}

Standard Pipeline

Upload The Champion Model

For our example, we will upload the champion model that has been trained to derive house prices from a variety of inputs. The model file is rf_model.onnx, and is uploaded with the name housingcontrol.

housing_model_control = (wl.upload_model(model_name_control, 
                                         model_file_name_control, 
                                         framework=wallaroo.framework.Framework.ONNX)
                                         .configure(tensor_fields=["tensor"])
                        )

Build the Pipeline

This pipeline is made to be an example of an existing situation where a model is deployed and being used for inferences in a production environment. We’ll call it housepricepipeline, set housingcontrol as a pipeline step, then run a few sample inferences.

mainpipeline = wl.build_pipeline(main_pipeline_name)
# in case this pipeline was run before
mainpipeline.clear()
mainpipeline.add_model_step(housing_model_control)

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

mainpipeline.deploy(deployment_config=deploy_config)

Testing

We’ll use two inferences as a quick sample test - one that has a house that should be determined around \$700k, the other with a house determined to be around \$1.5 million. We’ll also save the start and end periods for these events to for later log functionality.

dataframe_start = datetime.datetime.now()

normal_input = pd.DataFrame.from_records({"tensor": [
            [
                4.0, 
                2.5, 
                2900.0, 
                5505.0, 
                2.0, 
                0.0, 
                0.0, 
                3.0, 
                8.0, 
                2900.0, 
                0.0, 
                47.6063, 
                -122.02, 
                2970.0, 
                5251.0, 
                12.0, 
                0.0, 
                0.0
            ]
        ]
    }
)
result = mainpipeline.infer(normal_input)
display(result)

large_house_input = pd.DataFrame.from_records(
    {
        'tensor': [
            [
                4.0, 
                3.0, 
                3710.0, 
                20000.0, 
                2.0, 
                0.0, 
                2.0, 
                5.0, 
                10.0, 
                2760.0, 
                950.0, 
                47.6696, 
                -122.261, 
                3970.0, 
                20000.0, 
                79.0, 
                0.0, 
                0.0
            ]
        ]
    }
)
large_house_result = mainpipeline.infer(large_house_input)
display(large_house_result)

import time
time.sleep(10)
dataframe_end = datetime.datetime.now()

As one last sample, we’ll run through roughly 1,000 inferences at once and show a few of the results. For this example we’ll use an Apache Arrow table, which has a smaller file size compared to uploading a pandas DataFrame JSON file. The inference result is returned as an arrow table, which we’ll convert into a pandas DataFrame to display the first 20 results.

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

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

Standard Pipeline Logs

Pipeline logs with standard pipeline steps are retrieved either with:

• Pipeline logs which returns either a pandas DataFrame or Apache Arrow table.
• Pipeline export_logs which saves the logs either a pandas DataFrame JSON file or Apache Arrow table.

For full details, see the Wallaroo Documentation Pipeline Log Management guide.

Pipeline Log Method

The Pipeline logs method includes the following parameters. For a complete list, see the Wallaroo SDK Essentials Guide: Pipeline Log Management.

Pipeline Log Warnings

If the total number of logs the either the set limit or 10 MB in file size, the following warning is returned:

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

If the total number of logs requested either through the limit or through the start_datetime and end_datetime request is greater than 10 MB in size, the following error is displayed:

Warning: Pipeline log size limit exceeded. Only displaying 509 log messages. Please request a file using export_logs.

The following examples demonstrate displaying the logs, then displaying the logs between the control_model_start and control_model_end periods, then again retrieved as an Arrow table with the logs limited to only 5 entries.

# pipeline log retrieval - reverse chronological order

regular_logs = mainpipeline.logs()

display("Standard Logs")
display(len(regular_logs))
display(regular_logs)

# Display metadata

metadatalogs = mainpipeline.logs(dataset=["time", "out.variable", "metadata"])
display("Metadata Logs")
# Only showing the pipeline version for space reasons
display(metadatalogs.loc[:, ["time", "out.variable", "metadata.pipeline_version"]])

# Display logs restricted by date and limit 

display("Logs restricted by date")
arrow_logs = mainpipeline.logs(start_datetime=dataframe_start, end_datetime=dataframe_end, limit=50)

display(len(arrow_logs))
display(arrow_logs)

# # pipeline log retrieval limited to arrow tables
display(mainpipeline.logs(arrow=True))

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.
‘Standard Logs’
100

100 rows × 4 columns

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.
‘Metadata Logs’

100 rows × 3 columns

'Logs restricted by date'
2

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.
pyarrow.Table

time: timestamp[ms]

in.tensor: list<item: float> not null

child 0, item: float

out.variable: list<inner: float not null> not null

child 0, inner: float not null

anomaly.count: uint32 not null
time: [[2024-04-16 14:47:43.370,2024-04-16 14:47:43.370,2024-04-16 14:47:43.370,2024-04-16 14:47:43.370,2024-04-16 14:47:43.370,…,2024-04-16 14:47:43.370,2024-04-16 14:47:43.370,2024-04-16 14:47:43.370,2024-04-16 14:47:43.370,2024-04-16 14:47:43.370]]

in.tensor: [[[3,2,2005,7000,1,…,1750,4500,34,0,0],[3,1.75,2910,37461,1,…,2520,18295,47,0,0],…,[4,2.25,4470,60373,2,…,3210,40450,26,0,0],[3,1,1150,3000,1,…,1460,3200,108,0,0]]]

out.variable: [[[581003],[706823.56],…,[1208638],[448627.72]]]

anomaly.count: [[0,0,0,0,0,…,0,0,0,0,0]]

Standard Pipeline Steps Log Requests

Effected pipeline steps:

• add_model_step
• replace_with_model_step

For log file requests, the following metadata dataset requests for standard pipeline steps are available:

• metadata

These must be paired with specific columns. * is not available when paired with metadata.

• in: All input fields.
• out: All output fields.
• time: The DateTime the inference request was made.
• in.{input_fields}: Any input fields (tensor, etc.)
• out.{output_fields}: Any output fields (out.house_price, out.variable, etc.)
• anomaly.count: Any anomalies detected from validations.
• anomaly.{validation}: The validation that triggered the anomaly detection and whether it is True (indicating an anomaly was detected) or False.

The following requests the metadata, and displays the output variable and last model from the metadata.

# Display metadata

metadatalogs = mainpipeline.logs(dataset=['time', "out","metadata"])
display("Metadata Logs")
display(metadatalogs.loc[:, ['time', 'out.variable', 'metadata.last_model']])

Pipeline log schema has changed over the logs requested 1 newest records retrieved successfully, newest record seen was at <datetime>. Please request additional records separately
‘Metadata Logs’

Pipeline Limits

In a previous step we performed 10,000 inferences at once. If we attempt to pull them at once, we’ll likely run into the size limit for this pipeline and receive the following warning message indicating that the pipeline size limits were exceeded and we should use export_logs instead.

Warning: Pipeline log size limit exceeded. Only displaying 1000 log messages (of 10000 requested). Please request a file using export_logs.

logs = mainpipeline.logs(limit=10000)
display(logs)

Pipeline log schema has changed over the logs requested 1 newest records retrieved successfully, newest record seen was at <datetime>. Please request additional records separately

Pipeline export_logs Method

The Pipeline method export_logs returns the Pipeline records as either a DataFrame JSON file, or an Apache Arrow table file. For a complete list, see the Wallaroo SDK Essentials Guide: Pipeline Log Management.

The export_logs method takes the following parameters:

The following examples demonstrate saving a DataFrame version of the mainpipeline logs, then an Arrow version.

# Save the DataFrame version of the log file

mainpipeline.export_logs()
display(os.listdir('./logs'))

mainpipeline.export_logs(arrow=True)
display(os.listdir('./logs'))

Warning: There are more logs available. Please set a larger limit to export more data.
Note: The logs with different schemas are written to separate files in the provided directory.
[’logpipeline-1.arrow’,

’logpipeline-test-2.json’,

’logpipeline-1.json’,

’logpipeline-test-1.json']
Warning: There are more logs available. Please set a larger limit to export more data.
Note: The logs with different schemas are written to separate files in the provided directory.
[’logpipeline-1.arrow’,

’logpipeline-test-2.json’,

’logpipeline-1.json’,

’logpipeline-test-1.arrow’,

’logpipeline-test-2.arrow’,

’logpipeline-test-1.json’]

Shadow Deploy Pipelines

Let’s assume that after analyzing the assay information we want to test two challenger models to our control. We do that with the Shadow Deploy pipeline step.

In Shadow Deploy, the pipeline step is added with the add_shadow_deploy method, with the champion model listed first, then an array of challenger models after. All inference data is fed to all models, with the champion results displayed in the out.variable column, and the shadow results in the format out_{model name}.variable. For example, since we named our challenger models housingchallenger01 and housingchallenger02, the columns out_housingchallenger01.variable and out_housingchallenger02.variable have the shadow deployed model results.

For this example, we will remove the previous pipeline step, then replace it with a shadow deploy step with rf_model.onnx as our champion, and models xgb_model.onnx and gbr_model.onnx as the challengers. We’ll deploy the pipeline and prepare it for sample inferences.

# Upload the challenger models

model_name_challenger01 = 'logcontrolchallenger01'
model_file_name_challenger01 = './models/xgb_model.onnx'

model_name_challenger02 = 'logcontrolchallenger02'
model_file_name_challenger02 = './models/gbr_model.onnx'

housing_model_challenger01 = (wl.upload_model(model_name_challenger01, 
                                              model_file_name_challenger01, 
                                              framework=wallaroo.framework.Framework.ONNX)
                                              .configure(tensor_fields=["tensor"])
                            )
housing_model_challenger02 = (wl.upload_model(model_name_challenger02, 
                                              model_file_name_challenger02, 
                                              framework=wallaroo.framework.Framework.ONNX)
                                              .configure(tensor_fields=["tensor"])
                            )

# Undeploy the pipeline
mainpipeline.undeploy()

mainpipeline.clear()

# Add the new shadow deploy step with our challenger models
mainpipeline.add_shadow_deploy(housing_model_control, [housing_model_challenger01, housing_model_challenger02])

# Deploy the pipeline with the new shadow step
deploy_config = wallaroo.deployment_config.DeploymentConfigBuilder() \
    .cpus(0.25)\
    .build()

mainpipeline.deploy(deployment_config=deploy_config)

Shadow Deploy Sample Inference

We’ll now use our same sample data for an inference to our shadow deployed pipeline, then display the first 20 results with just the comparative outputs.

shadow_date_start = datetime.datetime.now()

shadow_result = mainpipeline.infer_from_file('./data/xtest-1k.arrow')

shadow_outputs =  shadow_result.to_pandas()
display(shadow_outputs.loc[0:20,['out.variable','out_logcontrolchallenger01.variable','out_logcontrolchallenger02.variable']])

shadow_date_end = datetime.datetime.now()

Shadow Deploy Logs

Pipelines with a shadow deployed step include the shadow inference result in the same format as the inference result: inference results from shadow deployed models are displayed as out_{model name}.{output variable}.

# display logs with shadow deployed steps

display(mainpipeline.logs(start_datetime=shadow_date_start, end_datetime=shadow_date_end).loc[:, ["time", "out.variable", "out_logcontrolchallenger01.variable", "out_logcontrolchallenger02.variable"]])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

1000 rows × 4 columns

For log file requests, the following metadata dataset requests for testing pipeline steps are available:

• metadata

These must be paired with specific columns. * is not available when paired with metadata.

• in: All input fields.
• out: All output fields.
• time: The DateTime the inference request was made.
• in.{input_fields}: Any input fields (tensor, etc.).
• out.{output_fields}: Any output fields matching the specific output_field (out.house_price, out.variable, etc.).
• out_: All shadow deployed challenger steps Any output fields matching the specific output_field (out.house_price, out.variable, etc.).
• anomaly.count: Any anomalies detected from validations.
• anomaly.{validation}: The validation that triggered the anomaly detection and whether it is True (indicating an anomaly was detected) or False.

The following example retrieves the logs from a pipeline with shadow deployed models, and displays the specific shadow deployed model outputs and the metadata.elasped field.

# display logs with shadow deployed steps

display(mainpipeline.logs(start_datetime=shadow_date_start, end_datetime=shadow_date_end).loc[:, ["time", 
                                                                                                  "out.variable", 
                                                                                                  "out_logcontrolchallenger01.variable", 
                                                                                                  "out_logcontrolchallenger02.variable"
                                                                                                  ]
                                                                                        ])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

1000 rows × 4 columns

metadatalogs = mainpipeline.logs(dataset=["time",
                                          "out_logcontrolchallenger01.variable", 
                                          "out_logcontrolchallenger02.variable", 
                                          "metadata",
                                          'anomaly.count'
                                          ],
                                start_datetime=shadow_date_start, 
                                end_datetime=shadow_date_end
                                )

display(metadatalogs.loc[:, ['out_logcontrolchallenger01.variable',	
                             'out_logcontrolchallenger02.variable', 
                             'metadata.elapsed',
                             'anomaly.count'
                             ]
                        ])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

1000 rows × 4 columns

The following demonstrates exporting the shadow deployed logs to the directory shadow.

# Save shadow deployed log files as pandas DataFrame

mainpipeline.export_logs(directory="shadow", file_prefix="shadowdeploylogs")
display(os.listdir('./shadow'))

Warning: There are more logs available. Please set a larger limit to export more data.
Note: The logs with different schemas are written to separate files in the provided directory.
[‘shadowdeploylogs-1.json’, ‘shadowdeploylogs-2.json’]

A/B Testing Pipeline

A/B testing allows inference requests to be split between a control model and one or more challenger models. For full details, see the Pipeline Management Guide: A/B Testing.

When the inference results and log entries are displayed, they include the column out._model_split which displays:

For this example, the shadow deployed step will be removed and replaced with an A/B Testing step with the ratio 1:1:1, so the control and each of the challenger models will be split randomly between inference requests. A set of sample inferences will be run, then the pipeline logs displayed.

pipeline = (wl.build_pipeline(“randomsplitpipeline-demo”)
.add_random_split([(2, control), (1, challenger)], “session_id”))

mainpipeline.undeploy()

# remove the shadow deploy steps
mainpipeline.clear()

# Add the a/b test step to the pipeline
mainpipeline.add_random_split([(1, housing_model_control), (1, housing_model_challenger01), (1, housing_model_challenger02)], "session_id")

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

mainpipeline.deploy(deployment_config=deploy_config)

# Perform sample inferences of 20 rows and display the results
ab_date_start = datetime.datetime.now()
abtesting_inputs = pd.read_json('./data/xtest-1k.df.json')

for index, row in abtesting_inputs.sample(20).iterrows():
    display(mainpipeline.infer(row.to_frame('tensor').reset_index()).loc[:,["out._model_split", "out.variable"]])

ab_date_end = datetime.datetime.now()

## Get the logs with the a/b testing information

metadatalogs = mainpipeline.logs(dataset=["time",
                                          "out", 
                                          "metadata"
                                          ]
                                )

display(metadatalogs.loc[:, ['out.variable', 'metadata.last_model']])

Pipeline log schema has changed over the logs requested 1 newest records retrieved successfully, newest record seen was at <datetime>. Please request additional records separately

# Save a/b testing log files as DataFrame

mainpipeline.export_logs(directory="abtesting", 
                         file_prefix="abtests", 
                         start_datetime=ab_date_start, 
                         end_datetime=ab_date_end)
display(os.listdir('./abtesting'))

['abtests-1.json']

The following exports the metadata with the log files.

# Save a/b testing log files as DataFrame

mainpipeline.export_logs(directory="abtesting-metadata", 
                         file_prefix="abtests", 
                         start_datetime=ab_date_start, 
                         end_datetime=ab_date_end,
                         dataset=["time", "out", "metadata"])
display(os.listdir('./abtesting-metadata'))

['abtests-1.json']

Anomaly Detection Logs

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 an error 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'))

Anomaly Detection Inference Requests Example

For this example, we create the validation rule too_high which detects houses with a value greater than 1,000,000 and show the output for houses that trigger that validation.

For these examples we’ll create a new pipeline to ensure the logs are “clean” for the samples.

import polars as pl

mainpipeline.undeploy()
newpipeline = wl.build_pipeline("logpipeline-anomaly-example")
newpipeline.clear()
newpipeline.add_model_step(housing_model_control)
newpipeline.add_validations(
    too_high=pl.col("out.variable").list.get(0) > 1000000.0
)

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

newpipeline.deploy(deployment_config=deploy_config)

import datetime
import time
import pytz

inference_start = datetime.datetime.now(pytz.utc)

# adding sleep to ensure log distinction
time.sleep(15)

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

inference_end = datetime.datetime.now(pytz.utc)

# first 20 results
display(results.head(20))

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

Anomaly Detection Logs

Pipeline logs retrieves with wallaroo.pipeline.logs include the anomaly dataset.

anomaly_logs =  newpipeline.logs(limit=1000)
display(anomaly_logs.loc[anomaly_logs['anomaly.too_high'] == True])

Warning: There are more logs available. Please set a larger limit or request a file using export_logs.

Undeploy Main Pipeline

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

mainpipeline.undeploy()