Do you have any Celery-like queue implementation?

[findalexli]

Hello!

I was drawn to this toolkit from your answer to Serve Deep learning Model with Celery or RESTful API?
https://stackoverflow.com/questions/58509111/serve-deep-learning-model-with-celery-or-restful-api

I am experimenting different framework's ability to construct a REST API with task queue as a single inference process for my project can take up an entire GPU's memory. I was trying to find if you have some Celery-like feature implemented.

[bojiang]

Hi, @findalexli.
BentoML V1.0 has a concept runner similar to the Celery task. BentoML not only queues the RESTful requests but also does automatic micro batching to use resources more efficiently.

[parano]

Adding to that, in the next release of Yatai, deployments created on kubernetes will be able to autoscale individual runners separately, which means for GPU/CPU/IO mixed workloads, the autoscaling will be targeting the bottle neck of your system and make more efficient use of your cluster resource.

For example, if your GPU based model have pre-processing, post-processing code that are CPU or IO intense and does not require GPU, BentoML/yatai will only run those code on non-GPU node in your cluster, and only the Model Runner that requires GPU will be assigned to GPU nodes.

[parano]

Another major feature in BentoML 1.0, is users can easily create complex inference graphs with multiple computation steps via the Runner API. It is very similar to what Celery does for batch oriented workload, but designed for real-time system.

Here is an example from the docs: https://docs.bentoml.org/en/latest/concepts/api_io_descriptors.html#sync-vs-async-apis

import aiohttp
import asyncio

# Load two runners for two different versions of the ScikitLearn
# Iris Classifier models we saved before
runner1 = bentoml.sklearn.load_runner("iris_classifier_model:yftvuwkbbbi6zcphca6rzl235")
runner2 = bentoml.sklearn.load_runner("iris_classifier_model:edq3adsfhzi6zgr6vtpeqaare")

# Create async API coroutine with pre-rocessing logic calling a feature store
@svc.api(input=NumpyNdarray(), output=NumpyNdarray())
async def predict(input_array: np.ndarray) -> np.ndarray:
# Call a remote feature store to pre-process the request
    async with aiohttp.ClientSession() as session:
    params = [("key", v) for v in a]
    async with session.get('https://features/get', params=input_array[0]) as resp:
        features = get_features(await resp.text())
    # Invoke both model runners simultaneously and return the better result
    results = await asyncio.gather(
        runner1.async_run(input_array, features),
        runner2.async_run(input_array, features),
    )
    return compare_results(results)