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

Wallaroo SDK Parallel Inference Demonstration: Aloha Model

This tutorial will focus on the Pipeline method parallel_infer, which allows a List of data to be submitted to a Wallaroo instance for parallel inference requests. This provides high speed increases in situations where data has to be broken up for size and memory needs, data is requested from multiple sources and submitted in a single request, or other use cases.

For this example we will be using an open source model that uses an Aloha CNN LSTM model for classifying Domain names as being either legitimate or being used for nefarious purposes such as malware distribution.

This example is a “toy” example. A proper use case would be in situations where the input data is multiple gigabytes in size, or where the model can only operate in batch mode.

Tutorial Goals

• Create a workspace for our work.
• Upload the Aloha TensorFlow model.
• Create a pipeline that can ingest our submitted data, submit it to the model, and export the results.
• Run a sample inference through our pipeline by loading a file.
• Run a batch inference to show submitting a set of data to an inference request.
• Perform parallel inference via two methods:
  • Split a DataFrame into a List:
    • Split a DataFrame into a List of 1,000 separate DataFrames to simulate separate inference requests.
    • Submit the List of DataFrames sequentially and display how long this takes.
  • Submit the same List of DataFrames with parallel_infer and compare how long it takes compared to the sequential list method.
  • Submit a DataFrame directly into the parallel_infer method and compare how long it takes compared to the sequential list method.

Prerequisites

• Wallaroo Version 2024.1 and above instance.

Reference

Wallaroo SDK Essentials Guide: Inference Management

Steps

Import Libraries

The first step will be to import our libraries.

import wallaroo
import asyncio 
import datetime

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

# to display dataframe tables
from IPython.display import display

import warnings
warnings.filterwarnings('ignore')

Open a Connection to Wallaroo

Next we connect to a Wallaroo instance 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.

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 names are unique across the Wallaroo instance. Verify either that another workspace with the same name does not exist, or that access is granted to the user.

workspace_name = f"alohaworkspace-parallel-infer"
pipeline_name = f"alohapipeline"
model_name = f"alohamodel"
model_file_name = './models/alohacnnlstm.zip'

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

wl.set_current_workspace(workspace)

aloha_pipeline = wl.build_pipeline(pipeline_name)
aloha_pipeline

Upload the Models

Now we will upload our models. Note that for this example we are applying the model from a .ZIP file. The Aloha model is a protobuf file that has been defined for evaluating web pages, and we will configure it to use data in the tensorflow format.

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

Deploy a model

Now that we have a model that we want to use we will create a deployment for it.

We will tell the deployment we are using a tensorflow model and give the deployment name and the configuration we want for the deployment.

aloha_pipeline.add_model_step(model)

REPLICAS = 2
deployment_config = (wallaroo.DeploymentConfigBuilder()
    .replica_count(REPLICAS)
    .build())

aloha_pipeline.deploy(deployment_config =deployment_config)

We can verify that the pipeline is running and list what models are associated with it.

aloha_pipeline.status()

{'status': 'Running',
 'details': [],
 'engines': [{'ip': '10.28.1.2',
   'name': 'engine-f9bd787d9-zz55j',
   'status': 'Running',
   'reason': None,
   'details': [],
   'pipeline_statuses': {'pipelines': [{'id': 'alohapipeline',
      'status': 'Running',
      'version': '0904f92a-1401-45f9-a6ef-aac8402daff5'}]},
   'model_statuses': {'models': [{'name': 'alohamodel',
      'sha': 'd71d9ffc61aaac58c2b1ed70a2db13d1416fb9d3f5b891e5e4e2e97180fe22f8',
      'status': 'Running',
      'version': 'fb36dcf3-9827-42e9-89c0-d607c668b2d6'}]}},
  {'ip': '10.28.5.5',
   'name': 'engine-f9bd787d9-d6xrg',
   'status': 'Running',
   'reason': None,
   'details': [],
   'pipeline_statuses': {'pipelines': [{'id': 'alohapipeline',
      'status': 'Running',
      'version': '0904f92a-1401-45f9-a6ef-aac8402daff5'}]},
   'model_statuses': {'models': [{'name': 'alohamodel',
      'sha': 'd71d9ffc61aaac58c2b1ed70a2db13d1416fb9d3f5b891e5e4e2e97180fe22f8',
      'status': 'Running',
      'version': 'fb36dcf3-9827-42e9-89c0-d607c668b2d6'}]}}],
 'engine_lbs': [{'ip': '10.28.1.4',
   'name': 'engine-lb-6b59985857-4vlzg',
   'status': 'Running',
   'reason': None,
   'details': []}],
 'sidekicks': []}

Sample Inferences

Infer 1 row

Now that the pipeline is deployed and our Aloha model is in place, we’ll perform a smoke test to verify the pipeline is up and running properly. We’ll use the infer_from_file command to load a single encoded URL into the inference engine and print the results back out.

The result should tell us that the tokenized URL is legitimate (0) or fraud (1) based on the out.main field. This sample data should return close to 0.

result = aloha_pipeline.infer_from_file('./data/data_1.df.json')

display(result)

Parallel Inference via the Wallaroo SDK

Parallel Inference Inference Request

The await wallaroo.pipeline.Pipeline.parallel_infer(tensor, timeout, num_parallel, retries) asynchronous method performs an inference as defined by the pipeline steps.

Parallel Inference Inference Request Parameters

Parallel Inference Inference Request Returns

wallaroo.pipeline.Pipeline.parallel_infer returns one of the following based on the tensor input parameter.

• If a pandas DataFrame was submitted, a pandas DataFrame is returned.
• If an Apache Arrow table was submitted, an Apache Arrow table is returned.

The following fields are returned based on the dataset and dataset_exclude parameters.

Parallel Inference Examples

Sequential Inference Example

The first example will show taking a pandas DataFrame with 25,000 rows, splitting those into 25,000 separate DataFrames with one row each, then submitting them sequentially. This simulates a situation where the data input sizes are so large they must be broken up for more efficient transmission and inferencing.

test_data = pd.read_json("./data/data_25k.df.json")
test_list = []

for index, row in test_data.head(1000).iterrows():
    test_list.append(row.to_frame('text_input').reset_index(drop=True))

#show the first row as an example
test_list[0]

#
# Run the inference sequentially to establish a baseline
#
now = datetime.datetime.now()

results = []
for df in test_list:
    results.append(aloha_pipeline.infer(tensor=df, timeout=10))

total_sequential = datetime.datetime.now() - now

print(f"Elapsed = {total_sequential.total_seconds()} : {len(results)}")

Elapsed = 240.318378 : 1000

Parallel Infer with DataFrame Example

The following example shows using wallaroo.pipeline.Pipeline.parallel_infer with a pandas DataFrames. It is automatically split for parallel inferences by the Wallaroo SDK.

We then compare it to the List of pandas DataFrames submitted sequentially.

timeout_secs=30
now = datetime.datetime.now()
parallel_results = await aloha_pipeline.parallel_infer(tensor=test_data.head(1000), 
                                                       timeout=timeout_secs, 
                                                       num_parallel=2*REPLICAS, 
                                                       retries=3)

total_parallel = datetime.datetime.now() - now
print(f"Elapsed_in_parallel = {total_parallel.total_seconds()} : {len(parallel_results)}")

Elapsed_in_parallel = 15.697349 : 1000

parallel_results.head(20)

Undeploy Pipeline

When finished with our tests, we will undeploy the pipeline so we have the Kubernetes resources back for other tasks. Note that if the deployment variable is unchanged aloha_pipeline.deploy() will restart the inference engine in the same configuration as before.

aloha_pipeline.undeploy()