"# Serve Ray AIR Predictors with `ModelWrapper`\n",
"\n",
"[Ray Serve](rayserve) is the recommended tool to deploy models trained with AIR.\n",
"\n",
"After training a model with Ray Train, you can serve a model using Ray Serve. In this guide, we will cover how to use Ray AIR's `ModelWrapper`, `Predictor`, and `Checkpoint` abstractions to quickly deploy a model for online inference.\n",
"\n",
"But before that, let's review the key concepts:\n",
"- [`Checkpoint`](ray.air.checkpoint) represents a trained model stored in memory, file, or remote uri.\n",
"- [`Predictor`](ray.air.predictor.Predictor)s understand how to perform a model inference given checkpoints and the model definition. Ray AIR comes with predictors for each supported frameworks. \n",
"- [`Deployment`](serve-key-concepts-deployment) is a Ray Serve construct that represent an HTTP endpoint along with scalable pool of models.\n",
"\n",
"The core concept for deployment is the `ModelWrapper`. The `ModelWrapper` takes a [predictor](ray.air.predictor.Predictor) class and a [checkpoint](ray.air.checkpoint) and transforms them into a live HTTP endpoint. \n",
"\n",
"We'll start with a simple quick-start demo showing how you can use the ModelWrapper to deploy your model for online inference."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"Let's first make sure Ray AIR is installed. For the quick-start, we'll also use Ray AIR to train and serve a XGBoost model."
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"!pip install \"ray[air]\" xgboost scikit-learn"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"You can find the preprocessor and trainer in the [key concepts walk-through](air-key-concepts)."
]
},
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"2022-06-02 19:31:31,356\tINFO services.py:1483 -- View the Ray dashboard at \u001b[1m\u001b[32mhttp://127.0.0.1:8265\u001b[39m\u001b[22m\n"
]
},
{
"data": {
"text/html": [
"== Status ==<br>Current time: 2022-06-02 19:31:48 (running for 00:00:13.38)<br>Memory usage on this node: 37.9/64.0 GiB<br>Using FIFO scheduling algorithm.<br>Resources requested: 0/16 CPUs, 0/0 GPUs, 0.0/25.71 GiB heap, 0.0/2.0 GiB objects<br>Result logdir: /Users/simonmo/ray_results/XGBoostTrainer_2022-06-02_19-31-34<br>Number of trials: 1/1 (1 TERMINATED)<br><table>\n",
"<thead>\n",
"<tr><th>Trial name </th><th>status </th><th>loc </th><th style=\"text-align: right;\"> iter</th><th style=\"text-align: right;\"> total time (s)</th><th style=\"text-align: right;\"> train-logloss</th><th style=\"text-align: right;\"> train-error</th><th style=\"text-align: right;\"> valid-logloss</th></tr>\n",
"\u001b[2m\u001b[36m(GBDTTrainable pid=60303)\u001b[0m UserWarning: `num_actors` in `ray_params` is smaller than 2 (1). XGBoost will NOT be distributed!\n",
"\u001b[2m\u001b[36m(GBDTTrainable pid=60303)\u001b[0m 2022-06-02 19:31:42,283\tINFO main.py:980 -- [RayXGBoost] Created 1 new actors (1 total actors). Waiting until actors are ready for training.\n",
"\u001b[2m\u001b[36m(GBDTTrainable pid=60303)\u001b[0m 2022-06-02 19:31:47,421\tINFO main.py:1519 -- [RayXGBoost] Finished XGBoost training on training data with total N=398 in 5.16 seconds (1.09 pure XGBoost training time).\n"
"The following block serves a Ray AIR model from a [checkpoint](ray.air.checkpoint), using the built-in [`XGBoostPredictor`](ray.air.predictors.integrations.xgboost.XGBoostPredictor)."
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"\u001b[2m\u001b[36m(ServeController pid=60981)\u001b[0m INFO 2022-06-02 19:31:52,825 controller 60981 checkpoint_path.py:17 - Using RayInternalKVStore for controller checkpoint and recovery.\n",
"\u001b[2m\u001b[36m(ServeController pid=60981)\u001b[0m INFO 2022-06-02 19:31:52,828 controller 60981 http_state.py:115 - Starting HTTP proxy with name 'SERVE_CONTROLLER_ACTOR:SERVE_PROXY_ACTOR-node:127.0.0.1-0' on node 'node:127.0.0.1-0' listening on '127.0.0.1:8000'\n",
"\u001b[2m\u001b[36m(HTTPProxyActor pid=60984)\u001b[0m INFO: Started server process [60984]\n",
"\u001b[2m\u001b[36m(ServeController pid=60981)\u001b[0m INFO 2022-06-02 19:31:55,191 controller 60981 deployment_state.py:1221 - Adding 1 replicas to deployment 'XGBoostService'.\n"
"It works! As you can see, you can use the `ModelWrapper` to deploy checkpoints trained in Ray AIR as live endpoints. You can find more end-to-end examples for your specific frameworks in the [examples](air-examples-ref) page.\n",
"\n",
"This tutorial aims to provide an in-depth understanding of `ModelWrappers`. In particular, it'll demonstrate:\n",
"- How to serve a predictor accepting array input.\n",
"- How to serve a predictor accepting dataframe input.\n",
"- How to serve a predictor accepting custom input that can be transformed to array or dataframe.\n",
"- How to configure micro-batching to enhance performance.\n",
"\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 1. Predictor accepting NumPy array\n",
"We'll use a simple predictor implementation that adds an increment to an input array."
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {},
"outputs": [],
"source": [
"import numpy as np\n",
"\n",
"from ray.air.predictor import Predictor\n",
"from ray.air.checkpoint import Checkpoint\n",
"\n",
"class AdderPredictor(Predictor):\n",
" \"\"\"Dummy predictor that increments input by a staic value.\"\"\"\n",
" def __init__(self, increment: int):\n",
" self.increment = increment\n",
" \n",
" @classmethod\n",
" def from_checkpoint(cls, ckpt: Checkpoint):\n",
" \"\"\"Create predictor from checkpoint.\n",
" \n",
" Args:\n",
" ckpt: The AIR checkpoint representing a single dictionary. The dictionary\n",
" should have key `increment` and an integer value.\n",
"It worked! Now let's serve it behind HTTP. In Ray Serve, the core unit of an HTTP service is called a [`Deployment`](serve-key-concepts-deployment). It turns a Python class into a queryable HTTP endpoint. For Ray AIR, Serve provides a `ModelWrapperDeployment` to simplify this transformation. You don't need to implement any Python classes. You just pass in your predictor and checkpoint instead.\n",
"\n",
"The deployment takes several arguments. It requires two arguments to start:\n",
"- `predictor_cls (Type[Predictor] | str)`: The predictor Python class. Typically you can use built-in integrations from Ray AIR like the `TorchPredictor`. Alternatively, you can specify the class path to import a predictor like `\"ray.air.integrations.torch.TorchPredictor\"`.\n",
"- `checkpoint (Checkpoint | str)`: A checkpoint instance, or uri to load the checkpoint from.\n",
"\n",
"The following cell showcases how to create a deployment with our `AdderPredictor`\n",
"\n",
"To learn more about Ray Serve, check out [its documentation](rayserve)."
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"\u001b[2m\u001b[36m(ServeController pid=60981)\u001b[0m INFO 2022-06-02 19:32:07,559 controller 60981 deployment_state.py:1221 - Adding 1 replicas to deployment 'Adder'.\n"
"Nice! We sent `[40]` as our input and got `[42]` as our output in JSON format.\n",
"\n",
"You can also specify multi-dimensional arrays in the JSON payload, as well as \"dtype\" and \"shape\" fields to process to array. For more information about the array input schema, see [Ndarray](serve-ndarray-schema).\n",
" \n",
"That's it for arrays! Let's take a look at tabular input."
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## 2. Predictor accepting Pandas DataFrame\n",
"Let's now take a look at a predictor accepting dataframe inputs. We'll perform some simple column-wise transformations on the input data."
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd\n",
"\n",
"\n",
"class DataFramePredictor(Predictor):\n",
" \"\"\"Dummy predictor that first multiplies input then increment it.\"\"\"\n",
"Just like the `AdderPredictor`, we'll use the same `ModelWrapperDeployment` approach to make it queryable with HTTP. \n",
"\n",
"Note that we added `http_adapter=pandas_read_json` as the keyword argument. This tells Serve how to convert incoming JSON requests into a DataFrame. The `pandas_read_json` adapter accepts:\n",
"- [Pandas-parsable JSON](https://pandas.pydata.org/docs/reference/api/pandas.read_json.html) in HTTP body\n",
"- Optional keyword arguments to the [`pandas.read_json`](https://pandas.pydata.org/docs/reference/api/pandas.read_json.html) function via HTTP url parameters.\n",
"\n",
"To learn more, see [HTTP Adapters](serve-http-adapters).\n",
"\n",
"```{note}\n",
"You might wonder why the previous array predictor doesn't need to specify any http adapter. This is because Ray Serve defaults to a built-in adapter called `json_to_ndarray`(ray.serve.http_adapters.json_to_ndarray)!\n",
"```"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"\u001b[2m\u001b[36m(ServeController pid=60981)\u001b[0m INFO 2022-06-02 19:32:24,396 controller 60981 deployment_state.py:1221 - Adding 1 replicas to deployment 'DataFramePredictor'.\n"
"Great! You can see that the input JSON has been converted to a dataframe, so our predictor can work with pure dataframes instead of raw HTTP requests.\n",
"\n",
"But what if we need to configure the HTTP request? You can do that as well.\n",
"\n",
"## 3. Accepting custom inputs via `http_adapter`\n",
"\n",
"The `http_adapter` field accepts any callable function that's type annotated. You can also bring in additional types that are accepted by FastAPI's dependency injection framework. For more detail, see [HTTP Adapters](serve-http-adapters). In the following example, instead of using the pandas adapter Serve provides, we'll implement our own request adapter that works with just http parameters instead of JSON."
"## 4. `ModelWrapper` performs microbatching to improve performance\n",
"\n",
"Common machine learning models take a batch of inputs for prediction. Common ML Frameworks are optimized with vectorized instruction to make inference on batch requests almost as fast as single requests. \n",
"\n",
"In Serve's `ModelWrapperDeployment`, the incoming requests are automatically batched. \n",
"\n",
"When multiple clients send requests at the same time, Serve will combine the requests into a single batch (array or dataframe). Then, Serve calls predict on the entire batch. Let's take a look at a predictor that returns each row's content, batch_size, and batch group."
]
},
{
"cell_type": "code",
"execution_count": 22,
"metadata": {},
"outputs": [],
"source": [
"import time\n",
"class BatchSizePredictor(Predictor):\n",
" @classmethod\n",
" def from_checkpoint(cls, _: Checkpoint):\n",
" return cls()\n",
" \n",
" def predict(self, inp: np.ndarray):\n",
" time.sleep(0.5) # simulate model inference.\n",
" return [(i, len(inp), inp) for i in inp]"
]
},
{
"cell_type": "code",
"execution_count": 23,
"metadata": {},
"outputs": [
{
"name": "stderr",
"output_type": "stream",
"text": [
"\u001b[2m\u001b[36m(ServeController pid=60981)\u001b[0m INFO 2022-06-02 19:33:39,305 controller 60981 deployment_state.py:1221 - Adding 1 replicas to deployment 'BatchSizePredictor'.\n"
