Model Deployments

With Chalk, you can deploy machine learning models as isolated services in dedicated scaling groups. Your models get their own compute resources, autoscaling policies, and lifecycle management, so you can scale and update them independently of the Chalk engine.

Model deployments host your models as standalone services that you can call from feature resolvers or external applications. You can use model inference in your feature pipeline while managing model serving separately from feature computation.

​

When to Use Model Deployments

Model deployments are ideal when you want to:

• Isolate model resources: Give models their own CPU, memory, and GPU resources independent of the engine
• Scale models independently: Auto-scale models based on inference demand without affecting other services
• Version and update models separately: Deploy new model versions without redeploying your entire Chalk system
• Run containerized models: Deploy models as Chalk images or Docker images without converting to Python objects
• Enable high-throughput inference: Run multiple replicas of your model in parallel

​

The @model_handler decorator

The fastest way to deploy a model is the @model_handler decorator. You write a single class with a predict method, hand a trained model object to register_model_version, and Chalk builds the serving image for you, ships your class as source, serializes and mounts the model, and wires up the runtime — no Dockerfile, no hand-written Arrow plumbing, no chalkcompute.Image.

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor

from chalk.client import ChalkClient
from chalk.ml import model_handler
from chalk.scalinggroup import ScalingGroupResourceRequest


@model_handler
class HousePriceModel:
    def predict(self, df):
        return pd.DataFrame({"price": self.model.predict(df.to_pandas())})


rng = np.random.default_rng(0)
X_train = rng.normal(size=(200, 2))  # columns: sqft, rooms
y_train = X_train @ [150.0, 50.0]    # price
rf = RandomForestRegressor().fit(X_train, y_train)

client = ChalkClient()
result = client.register_model_version(
    name="house_price",
    model=HousePriceModel(model=rf),
    input_schema={"sqft": float, "rooms": float},
    output_schema={"price": float},
    dependencies=["scikit-learn", "pandas", "chalkdf"],
)
deployment = client.create_model_deployment(
    name="house-price",
    model_name="house_price",
    model_version=result.model_version,
    resources=ScalingGroupResourceRequest(cpu="1", memory="2Gi"),
)

Pass input_schema, output_schema, and dependencies explicitly. Schemas are required — naming your columns is what makes the output line up with what predict returns — and dependencies is the list of pip packages your predict needs at runtime.

Schema values can be plain Python types — float, int, str, and bool map to pa.float64(), pa.int64(), pa.string(), and pa.bool_() respectively, so you don’t need to import pyarrow for ordinary tabular models. Reach for a PyArrow type only when you need one outside those four (for example pa.float32(), pa.large_string(), or a timestamp type).

The decorated class is a normal Python class with three Chalk-managed attributes injected:

• model: your trained model object. Chalk serializes it with the framework’s native serializer at registration time, uploads it to the model’s artifact volume, and deserializes it back into self.model inside the container before predict runs. The same attribute is available on both sides.
• files: a list of local file paths (files=["./scaler.pkl"]) that Chalk uploads to the artifact volume. At runtime self.files is a {basename: Path} mapping, so self.files["scaler.pkl"] resolves to the mounted path in the container (and to your local path when testing). Use this for tokenizers, scalers, encoders, lookup tables, etc.
• artifact_path: the mounted artifact volume directory.

​

predict and return types

predict(self, df) receives a chalkdf.DataFrame built from the request. Call df.to_pandas() (or df.to_arrow()) to get the shape your model expects. The return value can be any of the following — Chalk coerces it to the output columns for you:

Returning a named frame (pandas/polars/Arrow) is recommended so your output columns are explicit and match your output_schema.

​

Supported frameworks

The model= object can be any of: scikit-learn, PyTorch, XGBoost, LightGBM, CatBoost, TensorFlow/Keras, or ONNX. The framework is auto-detected and the right serializer is used.

​

Custom loading with load_model

For models Chalk can’t serialize (a custom Python class, a sentence encoder, a lookup table), leave model=None, ship the artifacts via files=[...], and own the loading in load_model, which runs once per replica at startup:

@model_handler
class CategoryEnricher:
    def load_model(self):
        import pickle
        with open(self.files["table.pkl"], "rb") as f:
            self.table = pickle.load(f)

    def predict(self, df):
        ids = df.to_pandas()["category_id"]
        return pd.DataFrame({"name": [self.table.get(i, "unknown") for i in ids]})

When you do have a model= object and also need extra setup (device placement, .eval(), auxiliary files), call self.default_load_model() inside your override to populate self.model the default way, then add your own steps:

@model_handler
class UserEmbedder:
    def load_model(self):
        self.default_load_model()                    # restores self.model
        self.model.to("cuda").eval()

​

Local testing

Because the decorated class is plain Python, you can exercise it locally with no container plumbing — construct it, call load_model() (a no-op outside the container when model= is already set), and call predict with a chalkdf.DataFrame:

import pyarrow as pa
from chalkdf import DataFrame

m = HousePriceModel(model=rf)
m.load_model()
out = m.predict(DataFrame.from_arrow(pa.RecordBatch.from_pydict({"sqft": [1000.0], "rooms": [3.0]})))

​

Examples by model type

Working @model_handler examples for each supported framework. The pattern is identical — write predict, hand over a trained model, register, deploy — but note that the object you get back as self.model in the container is whatever that framework’s loader returns, which is often not the class you trained (see the comments in each snippet). Switch tabs to see each one:

import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestRegressor

from chalk.client import ChalkClient
from chalk.ml import model_handler
from chalk.scalinggroup import ScalingGroupResourceRequest


@model_handler
class SklearnRF:
    def predict(self, df):
        X = df.to_pandas()[["f0", "f1", "f2", "f3"]]
        return pd.DataFrame({"prediction": self.model.predict(X)})


rng = np.random.default_rng(0)
X_train = rng.normal(size=(200, 4))
y_train = X_train @ [3.0, -2.0, 1.0, 0.5]
rf = RandomForestRegressor().fit(X_train, y_train)

client = ChalkClient()
v = client.register_model_version(
    name="sklearn_rf",
    model=SklearnRF(model=rf),
    input_schema={f"f{i}": float for i in range(4)},
    output_schema={"prediction": float},
    dependencies=["scikit-learn", "pandas", "chalkdf"],
)
deployment = client.create_model_deployment(
    name=f"sklearn-rf-{v.model_version}", model_name="sklearn_rf", model_version=v.model_version,
    resources=ScalingGroupResourceRequest(cpu="1", memory="2Gi"),
)

​

Container images

When you need full control over the serving container — a custom inference handler, extra system packages, or a pre-built image — register a model with a container image instead of a decorated class or a Python model object. Every model image runs the chalk-remote-call-python shim, which routes requests and handles PyArrow serialization. You supply a handler and an entrypoint that points chalk-remote-call at it, then register either a chalkcompute.Image (Chalk builds it) or a pre-built Docker image (you build it).

​

Writing a handler

The handler receives a dict of PyArrow Arrays — one per input_schema column — and returns a PyArrow Array. Optionally define on_startup to load resources once when the container starts; artifacts mounted via a volume live at /app/artifacts/ (see Automatic Volume Upload for Model Artifacts).

import json

import pyarrow as pa
import pyarrow.compute as pc

model = None


def on_startup():
    global model
    with open("/app/artifacts/model.json") as f:
        model = json.load(f)


def handler(event: dict[str, pa.Array], context: dict) -> pa.Array:
    factor = model["factor"]
    return pc.multiply(event["x"], pa.scalar(factor, type=pa.float64()))

Point the entrypoint at the handler (and optional startup hook):

chalk-remote-call --handler model.handler --on-startup model.on_startup --port 8080

​

Building a Chalk image

Pass a chalkcompute.Image and Chalk builds and manages the container for you. Install chalk-remote-call-python, add your handler file, and set chalk-remote-call as the entrypoint:

from chalk.client import ChalkClient
import chalkcompute

client = ChalkClient()

image = (
    chalkcompute.Image.debian_slim("3.11")
    .pip_install(["chalk-remote-call-python", "pyarrow"])
    .add_local_file("model.py", "/app/model.py", strategy="copy")
    .env({"PYTHONPATH": "/app"})
    .workdir("/app")
    .entrypoint(["chalk-remote-call", "--handler", "model.handler", "--port", "8080"])
)

client.register_model_version(
    name="my-model",
    input_schema={"x": float},
    output_schema={"y": float},
    model_image=image,
)

Useful chalkcompute.Image methods:

• .base(image): use a custom base Docker image
• .debian_slim(python_version): slim Debian base with the given Python version
• .pip_install(packages): install Python packages
• .run_commands(commands): run shell commands during the build
• .add_local_file(src, dest, strategy) / .add_local_dir(src, dest, strategy) — copy files in
• .env(vars) / .workdir(path) / .entrypoint(command) — set env vars, working dir, entrypoint

​

Using a pre-built Docker image

If you build and push the image yourself, register it by string reference. The image must meet the same requirements: install chalk-remote-call-python, define a handler (and optional on_startup), and run chalk-remote-call as the entrypoint.

client.register_model_version(
    name="my-model",
    input_schema={"x": float},
    output_schema={"y": float},
    model_image="my-model-image:latest",
)

FROM python:3.11-slim
WORKDIR /app
RUN pip install --no-cache-dir chalk-remote-call-python pyarrow
COPY model.py /app/model.py
ENV PYTHONPATH=/app
EXPOSE 8080
ENTRYPOINT ["chalk-remote-call", "--handler", "model.handler", "--port", "8080"]

docker build --platform linux/amd64 -t my-model:latest .
docker push my-model:latest

​

Automatic Volume Upload for Model Artifacts

When your model files are large (e.g. multi-gigabyte weight files), baking them into the container image is impractical — it slows down builds, increases image pull times, and wastes storage. Instead, Chalk automatically uploads model artifacts to a volume that gets mounted into your container at runtime. If your model artifacts are already baked into the image and you want to skip this automatic upload, pass skip_volume_upload=True during registration:

client.register_model_version(
    name="my-large-model",
    input_schema={"x": float},
    output_schema={"y": float},
    model_image="my-large-model-image:latest",
    skip_volume_upload=True,
)

The uploaded artifacts are mounted at /app/artifacts/ inside the container. Load them in on_startup exactly as shown in Writing a handler — open /app/artifacts/model.json once at startup and reference it from your handler.

​

Creating a model deployment

Use ChalkClient.create_model_deployment to deploy a registered model version. It returns a ModelDeployment with the deployment’s ID, name, selected revision_id, status, replica counts, and public web_url.

from chalk.client import ChalkClient
from chalk.scalinggroup import AutoScalingSpec, ScalingGroupResourceRequest

client = ChalkClient()

deployment = client.create_model_deployment(
    name="my-model-sg",
    model_name="my-model",
    model_version=1,
    handler="model.handler",
    scaling=AutoScalingSpec(
        min_replicas=1,
        max_replicas=2,
        target_cpu_utilization_percentage=70,
    ),
    resources=ScalingGroupResourceRequest(cpu="2", memory="4Gi"),
)
print(deployment.id, deployment.status, deployment.web_url)

Creation waits for readiness by default, with a wait_timeout of 300 seconds. Set wait_ready=False to return before the deployment is ready, then call deployment.wait_ready(timeout=600) when you need to wait. deployment.refresh() returns a new snapshot of the deployment’s state.

To change an existing deployment, use update_model_deployment. Creating with an existing name also records and selects a new revision of that deployment.

​

Auto-Scaling Configuration

Control how your model deployment scales based on demand using AutoScalingSpec.

from chalk.scalinggroup import AutoScalingSpec

# Configure auto-scaling behavior
scaling = AutoScalingSpec(
    min_replicas=1,                          # Minimum number of replicas
    max_replicas=5,                          # Maximum number of replicas
    target_cpu_utilization_percentage=70,    # Target CPU utilization (optional)
)

Chalk automatically scales the number of replicas based on inference request load and CPU utilization, staying within your min/max bounds. This ensures your models handle traffic spikes efficiently without wasting resources during quiet periods.

​

Resource Configuration

Specify CPU, memory, and GPU resources for each replica of your model using ScalingGroupResourceRequest.

from chalk.scalinggroup import ScalingGroupResourceRequest

# Request resources per replica
resources = ScalingGroupResourceRequest(
    cpu="2",                          # CPU allocation per replica
    memory="4Gi",                     # Memory allocation per replica
    gpu="nvidia-tesla-t4:1",          # Optional: GPU type and count
)

Each replica gets the specified resources. When Chalk scales from 1 to 3 replicas, total resource usage is multiplied accordingly (e.g., 3 replicas × 2 CPU = 6 CPU total).

​

Calling Deployed Models

You can call a deployed model three ways: directly from Python with remote() or defer()/get(), from a feature resolver with F.catalog_call, or from SQL. In every case the argument order must match your model’s input_schema.

​

Directly from Python: remote() and defer()/get()

Get a deployment by exactly one of its id or name, then call it directly:

from chalk.client import ChalkClient

client = ChalkClient()
deployment = client.get_model_deployment(name="house-price")

price = deployment.remote(sqft=1000.0, rooms=3.0)
price = deployment.remote(1000.0, 3.0)

handle = deployment.defer(sqft=1000.0, rooms=3.0)
price = handle.get(timeout=30)

remote() sends one row of feature values and returns the first output value when the model responds. Pass values by name or position, using the model’s input_features order for positional arguments.

Calls use the deployment’s selected revision, so updates and rollbacks apply to subsequent calls.

defer() enqueues the same inputs on the selected deployment’s stored queue and returns a ModelCallHandle immediately. Pending calls are consumed by the revision selected when they run. Older revisions can retain a shared, model-wide queue until redeployed; selecting an older deployment does not isolate its deferred calls from other consumers of that queue.

handle.get() raises ModelRemoteError if the call fails, or TimeoutError if its timeout expires before a result is available.

​

Gather multiple model calls

Submit several calls with defer(), then collect their results with chalk.gather():

from chalk import gather
from chalk.client import ChalkClient

deployment = ChalkClient().get_model_deployment(name="house-price")
handles = [
    deployment.defer(sqft=1000.0, rooms=3.0),
    deployment.defer(sqft=1500.0, rooms=4.0),
    deployment.defer(sqft=2000.0, rooms=5.0),
]
prices = gather(handles)

gather() waits for the results concurrently and returns a list in the same order as the handles. In async code, use prices = await gather(handles).

If any call fails, gather() raises the first exception it encounters without returning partial results. Other remote calls continue running.

​

Selecting a deployment through a model version

get_model_version also exposes remote() and defer() on a DeployedModelVersion. Select a deployment explicitly when a model version is served by several deployments:

model = client.get_model_version("house_price", deployment_name="house-price")
price = model.remote(sqft=1000.0, rooms=3.0)
print(model.deployment.id)

model = client.get_model_version(
    "house_price",
    version=model.version,
    deployment_id=model.deployment.id,
)

Use deployment_id or deployment_name to choose a deployment; version defaults to the version it serves. Without a deployment selector, get_model_version uses the requested version (or the latest registered version) and requires a single active deployment for inference.

​

From Resolvers

Address a model deployment by the name passed to create_model_deployment, referenced as model.{deployment_name}. Calls follow the deployment’s selected revision.

​

From a Feature Resolver — F.catalog_call

Call the model as part of feature computation, so its output is just another feature:

from chalk.features import features, _
from chalk import functions as F

@features
class MyModel:
    id: int
    x_1: float
    x_2: float
    y: float = F.catalog_call(
        "model.my-model-sg",
        _.x_1,
        _.x_2,
    )

F.catalog_call is evaluated during chalk query, so the call only takes effect once the feature graph is applied to your environment with chalk apply.

​

Asynchronous Calls — F.catalog_call_async

F.catalog_call_async takes the same arguments as F.catalog_call and produces the same result, but the engine enqueues the call on the scaling group and polls for the result instead of holding a blocking connection to the model server for the duration of the call. Use it for long-running or heavy inference, such as LLM generation or large batch scoring, where blocking calls would pin engine connections while the model works:

y: float = F.catalog_call_async(
    "model.my-model-sg",
    _.x_1,
    _.x_2,
)

​

From SQL Resolver — chalksql

In a SQL resolver, invoke the model with the catalog_call('model.{deployment_name}', ...) function, passing the model’s qualified name as the first argument:

select
    id,
    catalog_call('model.my-model-sg', x_1, x_2) as y
from my_table

​

Asynchronous SQL calls

catalog_call_async takes the same arguments as catalog_call and is the SQL counterpart of F.catalog_call_async: we enqueue and poll for a result instead of holding a connection open for the duration of the call. This is preferred for longer-running inference. Only model.* and function.* targets can be called like this.

select
    id,
    catalog_call_async('model.my-model-sg', x_1, x_2) as y
from my_table

​

Named arguments

After the call’s positional arguments, catalog_call and catalog_call_async accept named arguments that control how the call is made. They may be combined, and each may be given at most once.

select
    id,
    catalog_call_async(
        'model.my-model-sg', x_1, x_2,
        timeout => interval '90 seconds',
        on_error => 'struct'
    ) as y
from my_table

• timeout: how long a single request may take. Write it as an interval literal, or as a bare number of seconds (timeout => 90). Rows reach the model a batch at a time: one request per batch, and one request per distinct metadata value within it. Each of those requests gets the whole budget, so a slow request cannot spend another’s. A request that runs out of time fails; on_error decides what that does to the query. The deadline stops the query waiting, but does not abort the work: the model is not told, and finishes the request unread.
• on_error ('fail' / 'struct', default 'fail') — what a failed request does. 'fail' propagates it and fails the query, which is what catalog_call has always done. 'struct' reports it in the result instead: the call returns struct<value, error, started_at, ended_at>, where a successful row has its value and a null error, and a failed row has a null value and the failure’s message. Only the rows that failed are affected: with catalog_call_async that is the failing row alone, and otherwise the rows of the request that failed. Every other row, and the query itself, carry on. started_at and ended_at are UTC timestamps of when the row was sent (for catalog_call_async, when it was queued, so queue wait is included) and when its result or failure came back. Note that a model whose deployment configures retries does not retry under 'struct': the failure is reported rather than raised, so the retry policy never sees it.

-- Keep the rows the model answered, and see why the rest did not.
select
    id,
    y.value as score,
    y.error as failure
from (
    select id, catalog_call('model.my-model-sg', x_1, on_error => 'struct') as y
    from my_table
)

​

Serving an LLM with vLLM

create_model_deployment runs the image with the chalk-remote-call entrypoint (Chalk’s Arrow request/response runtime), so you can’t deploy a stock vLLM image directly — its own OpenAI HTTP server never starts. Instead, put the Chalk shim in front of vLLM: ship a handler that loads vLLM at startup and answers each Arrow request by generating with it. The model then deploys through the normal registration and model deployment path and is called like any other Chalk model (F.catalog_call or SQL). This is exactly how curated models are built.

import os

import pyarrow as pa

_llm = None


def on_startup():
    global _llm
    from vllm import LLM

    _llm = LLM(model=os.environ.get("VLLM_MODEL", "Qwen/Qwen2.5-0.5B-Instruct"))


def handler(event, context):
    from vllm import SamplingParams

    rb = pa.Table.from_pydict(event).combine_chunks().to_batches()[0]
    prompts = rb.column("prompt").to_pylist()
    outs = _llm.generate(prompts, SamplingParams(max_tokens=64))
    texts = [o.outputs[0].text for o in outs]
    return {"completion": pa.array(texts, type=pa.large_string())}

import pyarrow as pa
import chalkcompute

from chalk.client import ChalkClient
from chalk.scalinggroup import AutoScalingSpec, ScalingGroupResourceRequest

# vLLM base + the chalk-remote-call runtime + our handler.
image = (
    chalkcompute.Image.base("vllm/vllm-openai:latest")
    .pip_install(["chalk-remote-call-python", "pyarrow"])
    .add_local_file("handler.py", "/app/handler.py", strategy="copy")
    .workdir("/app")
    .env({"PYTHONPATH": "/app"})
)

client = ChalkClient()
result = client.register_model_version(
    name="vllm-qwen",
    model_image=image,
    input_schema={"prompt": pa.large_string()},
    output_schema={"completion": pa.large_string()},
)
deployment = client.create_model_deployment(
    name="vllm-qwen-sg",
    model_name="vllm-qwen",
    model_version=result.model_version,
    scaling=AutoScalingSpec(min_replicas=1, max_replicas=2),
    resources=ScalingGroupResourceRequest(gpu="nvidia-tesla-t4:1"),  # vLLM needs a GPU
    handler="handler.handler",
    env_vars={"PYTHONPATH": "/app"},
)

Then call it like any Chalk model — over the Arrow contract, not raw HTTP. For example, from a feature resolver:

from chalk.features import features, _
from chalk import functions as F

@features
class Doc:
    id: int
    prompt: str
    completion: str = F.catalog_call("model.vllm-qwen-sg", _.prompt)

​

Managing deployments

Use the model deployment APIs to inspect a deployment, select a new revision, or delete it.

​

Listing deployments

list_model_deployments returns one page. Filter by model_name and, optionally, model_version. Pass next_cursor back as cursor until it is None:

cursor = None
while True:
    page = client.list_model_deployments(model_name="my-model", cursor=cursor, limit=50)
    for item in page.deployments:
        print(item.id, item.name, item.model_version, item.status)
    cursor = page.next_cursor
    if cursor is None:
        break

​

Updating a deployment

Register the new model version, then replace the deployment’s serving spec with update_model_deployment.

import chalkcompute
from chalk.client.model_deployment import ModelDeploymentSpec

new_version = client.register_model_version(
    name="my-model",
    input_schema={"x": float},
    output_schema={"y": float},
    model_image=(
        chalkcompute.Image.debian_slim("3.11")
        .pip_install(["chalk-remote-call-python", "pyarrow"])
        .add_local_file("model_v2.py", "/app/model.py", strategy="copy")
        .env({"PYTHONPATH": "/app"})
        .workdir("/app")
        .entrypoint(["chalk-remote-call", "--handler", "model.handler", "--port", "8080"])
    ),
)

current = client.get_model_deployment(name="my-model-sg")
previous_revision_id = current.revision_id
deployment = client.update_model_deployment(
    id=current.id,
    spec=ModelDeploymentSpec(
        model_version=new_version.model_version,
        scaling=current.scaling,
        resources=current.resources,
        handler="model.handler",
    ),
)

ModelDeploymentSpec replaces the entire serving configuration. Supply its required model_version and scaling, and every optional setting you want to keep. Omitted resource requests, environment variables, secrets, probes, and workload identity take their defaults; they are not copied from the current revision. The example preserves the scaling and resources from the creation example. For a deployment with additional settings, include those settings from your deployment configuration as well.

Update and rollback wait for readiness by default, and accept wait_ready=False and wait_timeout just like creation.

​

Inspecting revisions and rolling back

list_model_deployment_revisions also returns one page at a time. Use next_cursor to read all revisions. Each revision has a stable id, a model_version, and a selected flag. After rollback, the selected revision can be older than the newest revision.

cursor = None
while True:
    page = client.list_model_deployment_revisions(
        deployment_id=deployment.id, cursor=cursor, limit=50
    )
    for revision in page.revisions:
        print(revision.id, revision.model_version, revision.selected)
    cursor = page.next_cursor
    if cursor is None:
        break

previous = client.get_model_deployment_revision(
    previous_revision_id, deployment_id=deployment.id
)
deployment = client.rollback_model_deployment(
    id=deployment.id,
    revision_id=previous.id,
)

Rollback selects an existing revision’s already-built spec without rebuilding its image. The endpoint and deferred calls follow the selected revision.

​

Deleting a deployment

Delete the serving deployment with delete_model_deployment:

deleted = client.delete_model_deployment(id=deployment.id)
archived = client.get_model_deployment(id=deleted.id, include_deleted=True)
print(archived.deleted_at)

Deletion removes the serving endpoint and leaves archived metadata available with include_deleted=True. That option is also supported by deployment lists and revision lookups/lists. Deleting a deployment does not delete the registered model version.

​

Structuring Your Model Deployment Code

Model registration and deployment should be controlled manually and separately from your feature definitions. Either:

1. Add to .chalkignore to prevent them from running during chalk apply.
2. Run in a separate repository dedicated to model management, keeping it independent from your Chalk feature code.

Your chalk apply will fail if it tries to run model registration and deployment code.

Organize your project to keep model management separate from feature definitions:

my-chalk-project/
|- models/                          # Model deployment code (add to .chalkignore)
|  |- model.py
|  `- deploy_model.py               # Registration + deployment script
|
|- features/                        # Feature definitions (synced with chalk apply)
|  |- __init__.py
|  `- user_features.py
|
|- .chalkignore
`- chalk.yaml

Put the following line in your .chalkignore so chalk apply skips everything under models/.

models/