This tutorial and the assets can be downloaded as part of the Wallaroo Tutorials repository.
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:
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.
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
pd.set_option('display.max_colwidth', None)
display(wallaroo.__version__)
'2023.2.1rc2'
Start a connect to the Wallaroo instance and save the connection into the variable wl
.
# Login through local Wallaroo instance
wl = wallaroo.Client()
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.
# used for unique connection names
import string
import random
suffix= ''.join(random.choice(string.ascii_lowercase) for i in range(4))
workspace_name = f'multiple-replica-forecast-tutorial-{suffix}'
pipeline_name = 'bikedaypipe'
model_name = 'bikedaymodel'
The workspace will be either used or created if it does not exist, along with the pipeline.
def get_workspace(name):
workspace = None
for ws in wl.list_workspaces():
if ws.name() == name:
workspace= ws
if(workspace == None):
workspace = wl.create_workspace(name)
return workspace
def get_pipeline(name):
try:
pipeline = wl.pipelines_by_name(name)[0]
except EntityNotFoundError:
pipeline = wl.build_pipeline(name)
return pipeline
workspace = get_workspace(workspace_name)
wl.set_current_workspace(workspace)
pipeline = get_pipeline(pipeline_name)
model_file_name = 'forecast.py'
bike_day_model = wl.upload_model(model_name, model_file_name, Framework.PYTHON).configure(runtime="python")
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
.
pipeline.add_model_step(bike_day_model)
name | bikedaypipe |
---|---|
created | 2023-07-14 15:50:50.014326+00:00 |
last_updated | 2023-07-14 15:50:52.029628+00:00 |
deployed | False |
tags | |
versions | 7aae4653-9e9f-468c-b266-4433be652313, 48983f9b-7c43-41fe-9688-df72a6aa55e9 |
steps | bikedaymodel |
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)
name | bikedaypipe |
---|---|
created | 2023-07-14 15:53:07.284131+00:00 |
last_updated | 2023-07-14 15:56:07.413409+00:00 |
deployed | True |
tags | |
versions | 9c67dd93-014c-4cc9-9b44-549829e613ad, 258dafaf-c272-4bda-881b-5998a4a9be26 |
steps | bikedaymodel |
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 cnt from bikerentals where date > DATE(DATE('2011-02-22'), '-1 month') AND date <= DATE('2011-02-22')
cnt | |
---|---|
0 | 986 |
1 | 1416 |
2 | 1985 |
3 | 506 |
4 | 431 |
pd.read_sql_query("select date, cnt from bikerentals where date > DATE(DATE('2011-02-22'), '-1 month') AND date <= DATE('2011-02-22') LIMIT 5", conn)
date | cnt | |
---|---|---|
0 | 2011-01-23 | 986 |
1 | 2011-01-24 | 1416 |
2 | 2011-01-25 | 1985 |
3 | 2011-01-26 | 506 |
4 | 2011-01-27 | 431 |
# send data to model for forecast
results = pipeline.infer(data.to_dict(orient='list'))[0]
results
{'forecast': [1462, 1483, 1497, 1507, 1513, 1518, 1521]}
# annotate with the appropriate dates (the next seven days)
resultframe = pd.DataFrame({
'date' : util.get_forecast_dates(first_day),
'forecast' : results['forecast']
})
# 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)
date | forecast | |
---|---|---|
0 | 2011-02-23 | 1462 |
1 | 2011-02-24 | 1483 |
2 | 2011-02-25 | 1497 |
3 | 2011-02-26 | 1507 |
4 | 2011-02-27 | 1513 |
5 | 2011-02-28 | 1518 |
6 | 2011-03-01 | 1521 |
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 cnt from bikerentals where date > DATE(DATE('2011-03-01'), '-1 month') AND date <= DATE('2011-03-01')
Current date: 2011-03-08
select cnt from bikerentals where date > DATE(DATE('2011-03-08'), '-1 month') AND date <= DATE('2011-03-08')
Current date: 2011-03-15
select cnt from bikerentals where date > DATE(DATE('2011-03-15'), '-1 month') AND date <= DATE('2011-03-15')
Current date: 2011-03-22
select cnt from bikerentals where date > DATE(DATE('2011-03-22'), '-1 month') AND date <= DATE('2011-03-22')
Current date: 2011-03-29
select cnt from bikerentals where date > DATE(DATE('2011-03-29'), '-1 month') AND date <= DATE('2011-03-29')
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:
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_list=inference_data, timeout=20, num_parallel=16, retries=2)
display(parallel_results)
[[{'forecast': [1764, 1749, 1743, 1741, 1740, 1740, 1740]}],
[{'forecast': [1735, 1858, 1755, 1841, 1770, 1829, 1780]}],
[{'forecast': [1878, 1851, 1858, 1856, 1857, 1856, 1856]}],
[{'forecast': [2363, 2316, 2277, 2243, 2215, 2192, 2172]}],
[{'forecast': [2225, 2133, 2113, 2109, 2108, 2108, 2108]}]]
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))
# upload to the database
for day_result in days_results:
resultframe = pd.DataFrame({
'date' : util.get_forecast_dates(day_result[0]),
'forecast' : day_result[1][0]['forecast']
})
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, cnt 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, cnt 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
date | forecast | actual | |
---|---|---|---|
0 | 2011-03-02 | 1764 | 2134 |
1 | 2011-03-03 | 1749 | 1685 |
2 | 2011-03-04 | 1743 | 1944 |
3 | 2011-03-05 | 1741 | 2077 |
4 | 2011-03-06 | 1740 | 605 |
5 | 2011-03-07 | 1740 | 1872 |
6 | 2011-03-08 | 1740 | 2133 |
7 | 2011-03-09 | 1735 | 1891 |
8 | 2011-03-10 | 1858 | 623 |
9 | 2011-03-11 | 1755 | 1977 |
10 | 2011-03-12 | 1841 | 2132 |
11 | 2011-03-13 | 1770 | 2417 |
12 | 2011-03-14 | 1829 | 2046 |
13 | 2011-03-15 | 1780 | 2056 |
14 | 2011-03-16 | 1878 | 2192 |
15 | 2011-03-17 | 1851 | 2744 |
16 | 2011-03-18 | 1858 | 3239 |
17 | 2011-03-19 | 1856 | 3117 |
18 | 2011-03-20 | 1857 | 2471 |
19 | 2011-03-21 | 1856 | 2077 |
20 | 2011-03-22 | 1856 | 2703 |
21 | 2011-03-23 | 2363 | 2121 |
22 | 2011-03-24 | 2316 | 1865 |
23 | 2011-03-25 | 2277 | 2210 |
24 | 2011-03-26 | 2243 | 2496 |
25 | 2011-03-27 | 2215 | 1693 |
26 | 2011-03-28 | 2192 | 2028 |
27 | 2011-03-29 | 2172 | 2425 |
28 | 2011-03-30 | 2225 | 1536 |
29 | 2011-03-31 | 2133 | 1685 |
Undeploy the pipeline and return the resources back to the Wallaroo instance.
conn.close()
pipeline.undeploy()
name | bikedaypipe |
---|---|
created | 2023-07-14 15:53:07.284131+00:00 |
last_updated | 2023-07-14 15:56:07.413409+00:00 |
deployed | False |
tags | |
versions | 9c67dd93-014c-4cc9-9b44-549829e613ad, 258dafaf-c272-4bda-881b-5998a4a9be26 |
steps | bikedaymodel |