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 a warning. Any validation rules named count are dropped upon request and an 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

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.

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

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

{'name': 'validation-house-price-demonstration', 'id': 9, 'archived': False, 'created_by': 'fb2916bc-551e-4a76-88e8-0f7d7720a0f9', 'created_at': '2024-07-29T20:21:47.038823+00:00', 'models': [{'name': 'anomaly-housing-model', 'versions': 1, 'owner_id': '""', 'last_update_time': datetime.datetime(2024, 7, 29, 20, 21, 48, 414367, tzinfo=tzutc()), 'created_at': datetime.datetime(2024, 7, 29, 20, 21, 48, 414367, tzinfo=tzutc())}], 'pipelines': [{'name': 'validation-demo', 'create_time': datetime.datetime(2024, 7, 29, 20, 21, 48, 755346, 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)

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': 'd340bfca-2152-4a4e-9375-1abebfbcb647', '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)

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)

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)

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