The following tutorials are available from the Wallaroo Tutorials Repository.

Stage 3: Deploy the Model in Wallaroo

In this stage, we upload the trained model and the processing steps to Wallaroo, then set up and deploy the inference pipeline.

Once deployed we can feed the newest batch of data to the pipeline, do the inferences and write the results to a results table.

For clarity in this demo, we have split the training/upload task into two notebooks:

• 02_automated_training_process.ipynb: Train and pickle ML model.
• 03_deploy_model.ipynb: Upload the model to Wallaroo and deploy into a pipeline.

Assuming no changes are made to the structure of the model, these two notebooks, or a script based on them, can then be scheduled to run on a regular basis, to refresh the model with more recent training data and update the inference pipeline.

This notebook is expected to run within the Wallaroo instance’s Jupyter Hub service to provide access to all required Wallaroo libraries and functionality.

Resources

The following resources are used as part of this tutorial:

• data
  • data/seattle_housing_col_description.txt: Describes the columns used as part data analysis.
  • data/seattle_housing.csv: Sample data of the Seattle, Washington housing market between 2014 and 2015.
• code
  • simdb.py: A simulated database to demonstrate sending and receiving queries.
• models
  • ./models/housing_model_xgb.onnx: Model created in Stage 2: Training Process Automation Setup.
  • ./models/preprocess_step.zip: A preprocessing model that formats the data for acceptance by the model.
  • ./models/postprocess_step.zip: A postprocessing model that formats the data output by housing_model_xgb.onnx for the inference output.

Steps

The process of uploading the model to Wallaroo follows these steps:

• Connect to Wallaroo: Connect to the Wallaroo instance and set up the workspace.
• Upload The Model: Upload the model and autoconvert for use in the Wallaroo engine.
• Upload the Processing Modules: Upload the processing modules.
• Create and Deploy the Pipeline: Create the pipeline with the model and processing modules as steps, then deploy it.
• Test the Pipeline: Verify that the pipeline works with the sample data.

Connect to Wallaroo

First we import the required libraries to connect to the Wallaroo instance, then connect to the Wallaroo instance.

import json
import pickle
import pandas as pd
import numpy as np
import pyarrow as pa

import simdb # module for the purpose of this demo to simulate pulling data from a database

# from wallaroo.ModelConversion import ConvertXGBoostArgs, ModelConversionSource, ModelConversionInputType
import wallaroo
from wallaroo.object import EntityNotFoundError

# 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

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

workspace_name = 'housepricing'
model_name = "housepricemodel"
model_file = "./models/housing_model_xgb.onnx"
pipeline_name = "housing-pipe"

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

Upload The Model

With the connection set and workspace prepared, upload the model created in 02_automated_training_process.ipynb into the current workspace.

To ensure the model input contract matches the provided input, the configuration tensor_fields=["tensor"] is used so regardless of what the model input type is, Wallaroo will ensure inputs of type tensor are accepted.

hpmodel = (wl.upload_model(model_name, 
                           model_file, 
                           framework=wallaroo.framework.Framework.ONNX)
                           .configure(tensor_fields=["tensor"]
                        )
            )

Upload the Processing Modules

We upload the preprocessing and postprocessing models that formats the incoming data into a shape the model is trained to accept, and the outgoing data from the model into a format we can use in our production system.

For more details on deploying Python models in Wallaroo, see Model Uploads and Registrations: Python Models.

input_schema = pa.schema([
    pa.field('id', pa.int64()),
    pa.field('date', pa.string()),
    pa.field('list_price', pa.float64()),
    pa.field('bedrooms', pa.int64()),
    pa.field('bathrooms', pa.float64()),
    pa.field('sqft_living', pa.int64()),
    pa.field('sqft_lot', pa.int64()),
    pa.field('floors', pa.float64()),
    pa.field('waterfront', pa.int64()),
    pa.field('view', pa.int64()),
    pa.field('condition', pa.int64()),
    pa.field('grade', pa.int64()),
    pa.field('sqft_above', pa.int64()),
    pa.field('sqft_basement', pa.int64()),
    pa.field('yr_built', pa.int64()),
    pa.field('yr_renovated', pa.int64()),
    pa.field('zipcode', pa.int64()),
    pa.field('lat', pa.float64()),
    pa.field('long', pa.float64()),
    pa.field('sqft_living15', pa.int64()),
    pa.field('sqft_lot15', pa.int64()),
    pa.field('sale_price', pa.float64())
])

output_schema = pa.schema([
    pa.field('tensor', pa.list_(pa.float32(), list_size=18))
])

preprocess_model = wl.upload_model("preprocess-step", "./models/preprocess_step.zip", \
                                   framework=wallaroo.framework.Framework.PYTHON, \
                                   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

Postprocess model:

input_schema = pa.schema([
    pa.field('variable', pa.list_(pa.float32()))
])

output_schema = pa.schema([
    pa.field('variable', pa.list_(pa.float32()))
])

postprocess_model = wl.upload_model("postprocess-step", "./models/postprocess_step.zip", \
                                   framework=wallaroo.framework.Framework.PYTHON, \
                                   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

Create and Deploy the Pipeline

Create the pipeline with the preprocess module, housing model, and postprocess module as pipeline steps, then deploy the newpipeline.

pipeline = wl.build_pipeline(pipeline_name)

pipeline.undeploy()
# clear if the tutorial was run before
pipeline.clear()

pipeline.add_model_step(preprocess_model)
pipeline.add_model_step(hpmodel)
pipeline.add_model_step(postprocess_model)

deploy_config = wallaroo.DeploymentConfigBuilder().replica_count(1).cpus(0.5).memory("1Gi").build()
pipeline.deploy()

pipeline.status()

{'status': 'Running',
 'details': [],
 'engines': [{'ip': '10.28.5.2',
   'name': 'engine-86c47f6568-n7xz2',
   'status': 'Running',
   'reason': None,
   'details': [],
   'pipeline_statuses': {'pipelines': [{'id': 'housing-pipe',
      'status': 'Running',
      'version': 'd610c4f2-5c05-41d6-b727-36c2dc21a04a'}]},
   'model_statuses': {'models': [{'name': 'postprocess-step',
      'sha': 'c4dfec3dd259395598646ce85b8efd7811840dc726bf4915c39d862b87fc7070',
      'status': 'Running',
      'version': 'eccf0ff4-1c63-4790-9957-a2553de50290'},
     {'name': 'preprocess-step',
      'sha': 'c09bbca6748ff23d83f48f57446c3ad6b5758c403936157ab731b3c269c0afb9',
      'status': 'Running',
      'version': '881fd6bb-c7bf-4917-9d4d-109e7c31a69e'},
     {'name': 'housepricemodel',
      'sha': 'd8b79e526eed180d39d4653b39bebd9d06e6ae7f68293b5745775a9093a3ae7d',
      'status': 'Running',
      'version': 'df316413-00ad-4b5f-ac46-5090524b0f78'}]}}],
 'engine_lbs': [{'ip': '10.28.5.4',
   'name': 'engine-lb-6b59985857-rhx8f',
   'status': 'Running',
   'reason': None,
   'details': []}],
 'sidekicks': [{'ip': '10.28.1.3',
   'name': 'engine-sidekick-postprocess-step-4-7df656b85c-jsq76',
   'status': 'Running',
   'reason': None,
   'details': [],
   'statuses': '\n'},
  {'ip': '10.28.5.3',
   'name': 'engine-sidekick-preprocess-step-3-55bd5d9c6f-cvcfd',
   'status': 'Running',
   'reason': None,
   'details': [],
   'statuses': '\n'}]}

Test the Pipeline

We will use a single query from the simulated housing_price table and infer. When successful, we will undeploy the pipeline to restore the resources back to the Kubernetes environment.

conn = simdb.simulate_db_connection()

# create the query
query = f"select * from {simdb.tablename} limit 1"
print(query)

# read in the data
singleton = pd.read_sql_query(query, conn)
conn.close()

display(singleton.loc[:, ["id", "date", "list_price", "bedrooms", "bathrooms", "sqft_living", "sqft_lot"]])

select * from house_listings limit 1

result = pipeline.infer(singleton)
# display the entire result
display(result)
# display just the output
display(result.loc[:, ['time', 'out.variable']])

1 rows × 25 columns

When finished, we undeploy the pipeline to return the resources back to the environment.

pipeline.undeploy()

 ok

With this stage complete, we can proceed to Stage 4: Regular Batch Inference.