Export data processing pipeline to torchscript for further deployment without Python environment#
Introduction#
When deploying models, we are more likely to use other languages (e.g. C++) instead of Python. This post provides a way by which users could save Chronos data processing pipeline as torchscript files to make it available without Python environment.
In this guide, we will
Develop a TCNForecaster with nyc_taxi dataset.
Export the data processing pipeline to torchscript after the forecaster is developed.
Show users how to use the saved pipeline when deploying the forecaster.
📝Note
Except exporting data processing pipeline to torchscript, you could also export the whole forecasting pipeline to torchscript. There is another how-to guide that illustrates this in detail, you may refer to it for more information.
Forecaster developing#
First let’s prepare the data. We will manually download the data to show the details.
[ ]:
# run following
!wget https://raw.githubusercontent.com/numenta/NAB/v1.0/data/realKnownCause/nyc_taxi.csv
Then we create a TSDataset
instance based on the data and do preprocessing. You could refer to
for details.
[ ]:
from sklearn.preprocessing import StandardScaler
from bigdl.chronos.data import TSDataset
import pandas as pd
# load the data to pandas dataframe
df = pd.read_csv("nyc_taxi.csv", parse_dates=["timestamp"])
# create TSDataset instance
train_data, _, test_data = TSDataset.from_pandas(df,
dt_col="timestamp",
target_col="value",
repair=False,
with_split=True,
test_ratio=0.1)
# create a scaler for data scaling
scaler = StandardScaler()
# preprocess train_data (scale and roll sampling)
train_data.scale(scaler, fit=True) \
.roll(lookback=48, horizon=24)
With the prepared data, we could easily develop a forecaster. You may refer to other how-to guides for more detail.
[ ]:
from bigdl.chronos.forecaster import TCNForecaster
# create a forecaster from tsdataset
forecaster = TCNForecaster.from_tsdataset(train_data)
# train the forecaster
forecaster.fit(train_data)
When a forecaster is developed with satisfying accuracy and performance, you could export the data processing pipeline to torchscript for further deployment. Currently preprocessing (including scale
and roll
) and postprocessing (including unscale_numpy
) can be exported by calling tsdataset.export_jit(path_dir=None, drop_dt_col=True)
, please check API
doc for more detailed information and current limitations.
[ ]:
from pathlib import Path
# create a directory to save the pipeline
saved_dir = Path("jit_module")
saved_dir.mkdir(exist_ok=True)
# export data processing pipeline to torchscript
train_data.export_jit(path_dir=saved_dir, drop_dt_col=True)
# save the test_data to csv files for deployment
# make sure to set `index=False` to make the saved data have the same structure as original data
test_data.df.to_csv("deployment_data.csv", index=False)
# export the forecaster to torchscript files for deployment
forecaster_path = saved_dir / "forecaster"
forecaster_path.mkdir(exist_ok=True)
forecaster.export_torchscript_file(dirname=forecaster_path, quantized_dirname=None)
Now the preprocessing pipeline and postprocessing pipeline are saved in jit_module/tsdata_preprocessing.pt
and jit_module/tsdata_postprocessing.pt
, you could load them when deploying the forecaster.
Forecaster deployment#
With the saved “.pt” files, you could do data preprocessing and postprocessing without Python environment. We provide a deployment workflow example in Python here since Python code can be directly executed in jupyter notebook, besides, a code snip representing the core deployment workflow in C++ using libtorch API is also presented below.
[ ]:
# deployment example in Python
import torch
# load the data from csv file
deployment_df = pd.read_csv("deployment_data.csv", parse_dates=["timestamp"])
# drop the datetime column because we specified `drop_dt_col=True` when exporting the pipeline
# now the data structure is same as data used in developing
deployment_df.drop(columns="timestamp", inplace=True)
# create input tensor
input_tensor = torch.from_numpy(deployment_df.values).type(torch.float64)
# load the saved pipelines
preprocess_path = saved_dir / "tsdata_preprocessing.pt"
postprocess_path = saved_dir / "tsdata_postprocessing.pt"
preprocess_module = torch.jit.load(preprocess_path)
postprocess_module = torch.jit.load(postprocess_path)
# preprocessing
preprocess_output = preprocess_module.forward(input_tensor)
# load the forecaster and inference
forecaster_module_path = forecaster_path / "ckpt.pth"
forecaster_module = torch.jit.load(forecaster_module_path)
inference_output = forecaster_module.forward(preprocess_output)
# postprocessing
postprocess_output = postprocess_module.forward(inference_output)
[ ]:
# Compare the result with the output of original deployment pipeline using Chronos API
from numpy.testing import assert_array_almost_equal
# preprocessing
test_data.scale(scaler, fit=False)\
.roll(lookback=48, horizon=24, is_predict=True)
input_data = test_data.to_numpy()
# inference
forecaster_module = torch.jit.load(forecaster_module_path)
inference_output_original = forecaster_module.forward(torch.from_numpy(input_data))
# postprocessing
postprocess_output_original = test_data.unscale_numpy(inference_output_original)
# compare the results
assert_array_almost_equal(postprocess_output.numpy(), postprocess_output_original)
Deployment in C++#
The following code describes the core deployment workflow in C++ using libtorch APIs. You could refer to installation guide to install libtorch, and more information of APIs is available at libtorch API doc.
// core deployment workflow example in C++
#include <torch/torch.h>
#include <torch/script.h>
// Create input tensor from your data, you should implement this function.
// The data to create input tensor should have the same format as the data used in developing.
// If you sepcified drop_dt_col=True when exporting the pipelines, you should skip the
// datatime column here to keep the same structure as the developing data.
torch::Tensor input_tensor = create_input_tensor(data);
// load the preprocessing pipeline
torch::jit::script::Module preprocessing;
preprocessing = torch::jit::load(preprocessing_path);
// run data preprocessing
torch::Tensor preprocessing_output = preprocessing.forward(input_tensor).toTensor();
// inference using your trained model, replace "trained_model" with your model
torch::Tensor inference_output = trained_model(preprocessing_output)
// load the postprocessing pipeline
torch::jit::script::Module postprocessing;
postprocessing = torch::jit::load(postprocessing_path);
// run postprocessing
torch::Tensor output = postprocessing.forward(inference_output).toTensor()