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.

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

How to preprocess my own data

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
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 specify the `drop_dt_col=True` when exporting the pipeline
# then 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()