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

Statsmodel Forecast with Wallaroo Features: Deploy and Test Infer

This tutorial series demonstrates how to use Wallaroo to create a Statsmodel forecasting model based on bike rentals. This tutorial series is broken down into the following:

• Create and Train the Model: This first notebook shows how the model is trained from existing data.
• Deploy and Sample Inference: With the model developed, we will deploy it into Wallaroo and perform a sample inference.
• Parallel Infer: A sample of multiple weeks of data will be retrieved and submitted as an asynchronous parallel inference. The results will be collected and uploaded to a sample database.
• External Connection: A sample data connection to Google BigQuery to retrieve input data and store the results in a table.
• ML Workload Orchestration: Take all of the previous steps and automate the request into a single Wallaroo ML Workload Orchestration.

In the previous step “Statsmodel Forecast with Wallaroo Features: Model Creation”, the statsmodel was trained and saved to the Python file forecast.py. This file will now be uploaded to a Wallaroo instance as a Python model, then used for sample inferences.

Prerequisites

• A Wallaroo instance version 2023.2.1 or greater.

References

• Wallaroo SDK Essentials Guide: Model Uploads and Registrations: Python Models
• Wallaroo SDK Essentials Guide: Pipeline Management
• Wallaroo SDK Essentials: Inference Guide: Parallel Inferences

Tutorial Steps

Import Libraries

The first step is to import the libraries that we will need.

import json
import os
import datetime
import pyarrow as pa

import wallaroo
from wallaroo.object import EntityNotFoundError
from wallaroo.framework import Framework

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

Initialize connection

Start a connect to the Wallaroo instance and save the connection into the variable wl.

# Login through local Wallaroo instance

wl = wallaroo.Client()

Set Configurations

The following will set the workspace, model name, and pipeline that will be used for this example. If the workspace or pipeline already exist, then they will assigned for use in this example. If they do not exist, they will be created based on the names listed below.

Workspace names must be unique. To allow this tutorial to run in the same Wallaroo instance for multiple users, the suffix variable is generated from a random set of 4 ASCII characters. To use the same workspace each time, hard code suffix and verify the workspace name created is is unique across the Wallaroo instance.

workspace_name = f'multiple-replica-forecast-tutorial'
pipeline_name = 'bikedaypipe'

Set the Workspace and Pipeline

The workspace will be either used or created if it does not exist, along with the pipeline.

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

wl.set_current_workspace(workspace)

pipeline = wl.build_pipeline(pipeline_name)

Upload Model

The Python model created in “Forecast and Parallel Infer with Statsmodel: Model Creation” will now be uploaded. Note that the Framework and runtime are set to python.

model_name = 'bikedaymodel'
model_file_name = './models/forecast_standard_new.zip'

input_schema = pa.schema([
    pa.field('count', pa.list_(pa.int32())) # time series to fit model
    ]
)

output_schema = pa.schema([
    pa.field('forecast', pa.list_(pa.int32())), # returns a forecast for a week (7 steps)
    pa.field('weekly_average', pa.float32()),
])

bike_day_model = wl.upload_model(model_name, 
                                 model_file_name, 
                                 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

bike_day_model

Deploy the Pipeline

We will now add the uploaded model as a step for the pipeline, then deploy it. The pipeline configuration will allow for multiple replicas of the pipeline to be deployed and spooled up in the cluster. Each pipeline replica will use 0.25 cpu and 512 Gi RAM.

pipeline = wl.build_pipeline(pipeline_name)

# Set the deployment to allow for additional engines to run
deploy_config = (wallaroo.DeploymentConfigBuilder()
                        .replica_count(1)
                        .replica_autoscale_min_max(minimum=2, maximum=5)
                        .cpus(0.25)
                        .memory("512Mi")
                        .build()
                    )

pipeline.add_model_step(bike_day_model).deploy(deployment_config=deploy_config)

pipeline.status()

{'status': 'Running',
 'details': [],
 'engines': [{'ip': '10.28.1.5',
   'name': 'engine-5b588dff86-fdqdp',
   'status': 'Running',
   'reason': None,
   'details': [],
   'pipeline_statuses': {'pipelines': [{'id': 'bikedaypipe',
      'status': 'Running',
      'version': 'd9a483b9-8b85-44d2-a556-c0cabf3285ab'}]},
   'model_statuses': {'models': [{'name': 'bikedaymodel',
      'sha': '96b4b27039f697f8a36ad15481e2d318cf603995553200b553c53f87a254fb2c',
      'status': 'Running',
      'version': '2462b419-0477-43c5-99ad-ce4f013561c0'}]}},
  {'ip': '10.28.1.7',
   'name': 'engine-5b588dff86-vhwbs',
   'status': 'Running',
   'reason': None,
   'details': [],
   'pipeline_statuses': {'pipelines': [{'id': 'bikedaypipe',
      'status': 'Running',
      'version': 'd9a483b9-8b85-44d2-a556-c0cabf3285ab'}]},
   'model_statuses': {'models': [{'name': 'bikedaymodel',
      'sha': '96b4b27039f697f8a36ad15481e2d318cf603995553200b553c53f87a254fb2c',
      'status': 'Running',
      'version': '2462b419-0477-43c5-99ad-ce4f013561c0'}]}}],
 'engine_lbs': [{'ip': '10.28.1.6',
   'name': 'engine-lb-6b59985857-gw2pt',
   'status': 'Running',
   'reason': None,
   'details': []}],
 'sidekicks': [{'ip': '10.28.1.3',
   'name': 'engine-sidekick-bikedaymodel-18-74df7478bf-x5fqq',
   'status': 'Running',
   'reason': None,
   'details': [],
   'statuses': '\n'},
  {'ip': '10.28.1.2',
   'name': 'engine-sidekick-bikedaymodel-18-74df7478bf-q42tc',
   'status': 'Running',
   'reason': None,
   'details': [],
   'statuses': '\n'}]}

Run Inference

Run a test inference to verify the pipeline is operational from the sample test data stored in ./data/testdata_dict.json.

from resources import simdb as simdb

def mk_dt_range_query(*, tablename: str, seed_day: str) -> str:
    assert isinstance(tablename, str)
    assert isinstance(seed_day, str)
    query = f"select count from {tablename} where date > DATE(DATE('{seed_day}'), '-1 month') AND date <= DATE('{seed_day}')"
    return query

conn = simdb.get_db_connection()

# create the query
query = mk_dt_range_query(tablename=simdb.tablename, seed_day='2011-03-01')
print(query)

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

select count from bikerentals where date > DATE(DATE('2011-03-01'), '-1 month') AND date <= DATE('2011-03-01')

data = {
        'count': [training_frame['count'].tolist()]
}
df = pd.DataFrame(data)
df

results = pipeline.infer(df)
display(results)

Undeploy the Pipeline

Undeploy the pipeline and return the resources back to the Wallaroo instance.

pipeline.undeploy()