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

Statsmodel Forecast with Wallaroo Features: Parallel Inference and Simulated Database Inferences

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.

This step will use the simulated database simdb to gather 4 weeks of inference data, then submit the inference request through the asynchronous Pipeline method parallel_infer. This receives a List of inference data, submits it to the Wallaroo pipeline, then receives the results as a separate list with each inference matched to the input submitted.

The results are then compared against the actual data to see if the model was accurate.

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

Parallel Infer Steps

Import Libraries

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

import json
import os
import datetime

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
import numpy as np

from resources import simdb
from resources import util

import importlib
importlib.reload(util)

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 across the tutorial notebooks, 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'
model_name = 'bikedaymodel'

Set the Workspace and Pipeline

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

The pipeline was configured in the previous step with the forecast model set as a pipeline step, so we don’t need to reconfigure it.

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

wl.set_current_workspace(workspace)

pipeline = wl.get_pipeline(pipeline_name)
pipeline

We already uploaded the model in the previous notebook, so we will retrieve the most recent version of that model with wallaroo.client.Client.get_model(model_name) and use that in our demonstrations.

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.

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

pipeline.deploy(deployment_config = deploy_config)

pipeline.status()

{'status': 'Running',
 'details': [],
 'engines': [{'ip': '10.28.1.14',
   'name': 'engine-85c5f89ff5-dc2vd',
   'status': 'Running',
   'reason': None,
   'details': [],
   'pipeline_statuses': {'pipelines': [{'id': 'bikedaypipe',
      'status': 'Running',
      'version': 'f0248c9e-8ce3-4038-8457-b48dff08adcf'}]},
   'model_statuses': {'models': [{'name': 'bikedaymodel',
      'sha': '96b4b27039f697f8a36ad15481e2d318cf603995553200b553c53f87a254fb2c',
      'status': 'Running',
      'version': '2462b419-0477-43c5-99ad-ce4f013561c0'}]}}],
 'engine_lbs': [{'ip': '10.28.1.11',
   'name': 'engine-lb-6b59985857-hffrq',
   'status': 'Running',
   'reason': None,
   'details': []}],
 'sidekicks': [{'ip': '10.28.1.12',
   'name': 'engine-sidekick-bikedaymodel-18-74df7478bf-d4dhw',
   'status': 'Running',
   'reason': None,
   'details': [],
   'statuses': '\n'}]}

Run Inference

For this example, we will forecast bike rentals by looking back one month from “today” which will be set as 2011-02-22. The data from 2011-01-23 to 2011-01-27 (the 5 days starting from one month back) are used to generate a forecast for what bike sales will be over the next week from “today”, which will be 2011-02-23 to 2011-03-01.

# retrieve forecast schedule
first_day, analysis_days = util.get_forecast_days()

print(f'Running analysis on {first_day}')

Running analysis on 2011-02-22

# connect to SQL data base 
conn = simdb.get_db_connection()
print(f'Bike rentals table: {simdb.tablename}')

# create the query and retrieve data
query = util.mk_dt_range_query(tablename=simdb.tablename, forecast_day=first_day)
print(query)
data = pd.read_sql_query(query, conn)
data.head()

Bike rentals table: bikerentals
select count from bikerentals where date > DATE(DATE('2011-02-22'), '-1 month') AND date <= DATE('2011-02-22')

pd.read_sql_query("select date, count from bikerentals where date > DATE(DATE('2011-02-22'), '-1 month') AND date <= DATE('2011-02-22') LIMIT 5", conn)

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

# send data to model for forecast

results = pipeline.infer(df)
results

util.get_forecast_dates(first_day).to_list()

['2011-02-23',
 '2011-02-24',
 '2011-02-25',
 '2011-02-26',
 '2011-02-27',
 '2011-02-28',
 '2011-03-01']

# annotate with the appropriate dates (the next seven days)
resultframe = pd.DataFrame({
     'date': util.get_forecast_dates(first_day).to_list(),
     'forecast' : results['out.forecast'][0]
})

# # write the new data to the db table "bikeforecast"
resultframe.to_sql('bikeforecast', conn, index=False, if_exists='append')

# # display the db table
query = "select date, forecast from bikeforecast"
pd.read_sql_query(query, conn)

Four Weeks of Inference Data

Now we’ll go back staring at the “current data” of 2011-03-01, and fetch each week’s data across the month. This will be used to submit 5 inference requests through the Pipeline parallel_infer method.

The inference data is saved into the inference_data List - each element in the list will be a separate inference request.

# get our list of items to run through

inference_data = []

content_type = "application/json"

days = []

for day in analysis_days:
    print(f"Current date: {day}")
    days.append(day)
    query = util.mk_dt_range_query(tablename=simdb.tablename, forecast_day=day)
    print(query)
    data = pd.read_sql_query(query, conn)
    inference_data.append(data.to_dict(orient='list'))

Current date: 2011-03-01
select count from bikerentals where date > DATE(DATE('2011-03-01'), '-1 month') AND date <= DATE('2011-03-01')
Current date: 2011-03-08
select count from bikerentals where date > DATE(DATE('2011-03-08'), '-1 month') AND date <= DATE('2011-03-08')
Current date: 2011-03-15
select count from bikerentals where date > DATE(DATE('2011-03-15'), '-1 month') AND date <= DATE('2011-03-15')
Current date: 2011-03-22
select count from bikerentals where date > DATE(DATE('2011-03-22'), '-1 month') AND date <= DATE('2011-03-22')
Current date: 2011-03-29
select count from bikerentals where date > DATE(DATE('2011-03-29'), '-1 month') AND date <= DATE('2011-03-29')

analysis_days

['2011-03-01', '2011-03-08', '2011-03-15', '2011-03-22', '2011-03-29']

df2 = pd.DataFrame(inference_data)
df2

Parallel Inference Request

The List inference_data will be submitted. Recall that the pipeline deployment can spool up to 5 replicas.

The pipeline parallel_infer(tensor_list, timeout, num_parallel, retries) asynchronous method performs an inference as defined by the pipeline steps and takes the following arguments:

• tensor_list (REQUIRED List): The data submitted to the pipeline for inference as a List of the supported data types:
  • pandas.DataFrame: Data submitted as a pandas DataFrame are returned as a pandas DataFrame. For models that output one column based on the models outputs.
  • Apache Arrow (Preferred): Data submitted as an Apache Arrow are returned as an Apache Arrow.
• timeout (OPTIONAL int): A timeout in seconds before the inference throws an exception. The default is 15 second per call to accommodate large, complex models. Note that for a batch inference, this is per list item - with 10 inference requests, each would have a default timeout of 15 seconds.
• num_parallel (OPTIONAL int): The number of parallel threads used for the submission. This should be no more than four times the number of pipeline replicas.
• retries (OPTIONAL int): The number of retries per inference request submitted.

parallel_infer is an asynchronous method that returns the Python callback list of tasks. Calling parallel_infer should be called with the await keyword to retrieve the callback results.

For more details, see the Wallaroo parallel inferences guide.

parallel_results = await pipeline.parallel_infer(tensor=df2, timeout=20, num_parallel=16, retries=2)

display(parallel_results)

Upload into DataBase

With our results, we’ll merge the results we have into the days we were looking to analyze. Then we can upload the results into the sample database and display the results.

# merge the days and the results

days_results = list(zip(days, parallel_results['out.forecast'].values))

# upload to the database
for day_result in days_results:
    # display(day_result[0])
    resultframe = pd.DataFrame({
        'date' : util.get_forecast_dates(day_result[0]),
        'forecast' : day_result[1][0]
    })
    display(resultframe)
    resultframe.to_sql('bikeforecast', conn, index=False, if_exists='append')

On April 1st, we can compare March forecasts to actuals

query = f'''SELECT bikeforecast.date AS date, forecast, count AS actual
            FROM bikeforecast LEFT JOIN bikerentals
            ON bikeforecast.date = bikerentals.date
            WHERE bikeforecast.date >= DATE('2011-03-01')
            AND bikeforecast.date <  DATE('2011-04-01')
            ORDER BY 1'''

print(query)

comparison = pd.read_sql_query(query, conn)
comparison

SELECT bikeforecast.date AS date, forecast, count AS actual
            FROM bikeforecast LEFT JOIN bikerentals
            ON bikeforecast.date = bikerentals.date
            WHERE bikeforecast.date >= DATE('2011-03-01')
            AND bikeforecast.date <  DATE('2011-04-01')
            ORDER BY 1

Undeploy the Pipeline

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

conn.close()
pipeline.undeploy()