diff --git a/docs/user_guides/fs/feature_group/delta_maintenance.md b/docs/user_guides/fs/feature_group/delta_maintenance.md new file mode 100644 index 0000000000..0dfcffb368 --- /dev/null +++ b/docs/user_guides/fs/feature_group/delta_maintenance.md @@ -0,0 +1,90 @@ +--- +description: Documentation on compacting, checkpointing and vacuuming Delta Feature Groups in Hopsworks. +--- + +# How to maintain a Delta Feature Group { #delta-maintenance-feature-group } + +## Introduction + +A Delta table that is written to repeatedly accumulates two things: data files and log entries. +Every commit writes at least one new data file, and every reader opens all of them. +Every commit also appends to the `_delta_log`, and a reader replays that log from the last checkpoint. +Neither is reclaimed on its own, and on a table written from Python neither is bounded on its own either: Spark writes a checkpoint every `delta.checkpointInterval` commits, delta-rs writes none. + +Four methods on a feature group bound them. +They apply only to feature groups with `time_travel_format="DELTA"` and return `None` for any other format. + +| Method | What it does | +| --- | --- | +| `delta_optimize` | Rewrites many small files into fewer large ones. Also available as `delta_compact`. | +| `delta_checkpoint` | Writes a checkpoint, so readers stop replaying the log from commit zero. | +| `delta_cleanup_metadata` | Expires the log entries a checkpoint already covers. | +| [`delta_vacuum`][hsfs.feature_group.FeatureGroup.delta_vacuum] | Deletes the data files no retained version references. | + +Each dispatches on the engine, so the same call works from a Python client with +delta-rs and from a PySpark job with Delta Spark. The first three are rendered as +plain code rather than API links until the client release that ships them, because +the docs build resolves cross-references against the released client. + +## Prerequisites + +Before you begin this guide we suggest you read the [Feature Group](../../../concepts/fs/feature_group/fg_overview.md) concept page and the [create feature group][create-feature-group] guide. + +## The maintenance sequence + +Run them in this order. + +```python +fg = fs.get_feature_group("transactions", version=1) + +fg.delta_optimize(max_concurrent_tasks=1) +fg.delta_checkpoint() +fg.delta_cleanup_metadata() +fg.delta_vacuum(retention_hours=168) +``` + +The order is what makes each step safe. +Compaction replaces many small files with few large ones and leaves the old ones on disk, still referenced by older versions. +The checkpoint goes next, so the smaller file list is recorded before anything is deleted. +Only then the two deletions: the log entries the checkpoint now covers, and the data files the compaction orphaned. + +## Choosing a retention + +`delta_vacuum` deletes files that versions inside the retention window no longer reference. +A query that is already running holds no lock on those files, so the retention has to stay comfortably longer than the longest query that runs against the group. +It is also the time travel window: a version whose files have been vacuumed cannot be read, which is why a compaction has to be followed by a checkpoint. + +The effect of a short retention is not that a vacuum deletes more, but that it deletes sooner. +A run reclaims what earlier runs orphaned rather than its own rewrite, whose files are seconds old. + +!!! warning "Delta's own floor" + Delta refuses a retention under seven days unless its retention check is disabled. + Hopsworks disables that check for you so a shorter retention takes effect, which means the value you pass is the value that applies. + Pick it against your own readers rather than relying on the engine to refuse a bad one. + +## Compacting only what changed + +On a table partitioned by a date column, `after_ingest_date` bounds the rewrite to partitions at or after that date. + +```python +fg.delta_optimize(after_ingest_date="2026-09-10") +``` + +Use it for anything that runs on a schedule. +Only files written since the last compaction need rewriting, and on a date-partitioned table they are all at or after that date, so bounding the rewrite this way keeps its cost flat. +Without it every run rewrites the whole table, including everything earlier runs already compacted, and the cost grows with the table forever. +Leave a day of slack for rows that arrived late. + +Only a partition column can select files without reading them, so this is refused on a group that is not partitioned by a date. +Compact the whole table by leaving `after_ingest_date` unset. + +## When to run them + +For an append-heavy table, compact when the active file count crosses a threshold and otherwise once a day. +Around 100 files is the low hundreds of megabytes at typical commit sizes, near the engine's own target file size. + +Read the last compaction time from the table's own history rather than keeping state, so the schedule survives restarts and multiple writers. + +These can run from a [Hopsworks job](../../projects/jobs/pyspark_job.md) on a schedule. +Compaction is the only one of the four that a deployment reading the same table notices: measured beside live traffic it roughly doubled p99 for the few seconds it ran, while the median moved by a tenth of a millisecond. +The other three sat where the deployment sat with nothing running. diff --git a/docs/user_guides/fs/feature_view/feature-vectors.md b/docs/user_guides/fs/feature_view/feature-vectors.md index 125e3615b5..114406aa5a 100644 --- a/docs/user_guides/fs/feature_view/feature-vectors.md +++ b/docs/user_guides/fs/feature_view/feature-vectors.md @@ -271,6 +271,31 @@ After [defining a transformation function using a context variable](../transform ) ``` +## Retrieving feature vectors without blocking + +`get_feature_vector` and `get_feature_vectors` block the calling thread for the whole round trip to the online store. +Inside a serving deployment, or anywhere else that runs an event loop, that stops every other request while the lookup is in flight. +`get_feature_vector_async` and `get_feature_vectors_async` take the same arguments and return the same values, awaited instead. + +```python +vector = await my_feature_view.get_feature_vector_async(entry={"pk1": 1, "pk2": 2}) + +vectors = await my_feature_view.get_feature_vectors_async( + entry=[{"pk1": 1, "pk2": 2}, {"pk1": 3, "pk2": 4}] +) +``` + +The statements are awaited on the caller's own event loop, against a connection pool belonging to that loop, so several lookups are in flight at once. +On a measured deployment this raised throughput from 218 to 270 requests per second and cut p99 latency by 72 percent. + +The awaited path applies to the SQL client. +A deployment reading through the REST client falls back to the blocking call, since there is nothing there to overlap. + +Each event loop gets its own connection pool, and that pool is released when its loop is collected. +A process that creates a loop per lookup, for example by calling `asyncio.run` in a loop, therefore does not accumulate connections that way. + +The default predictor a deployment gets from `model.deploy()` or `feature_view.deploy()` already awaits its lookup. + ## Choose the right Client The Online Store can be accessed via the **Python** or **Java** client allowing you to use your language of choice to connect to the Online Store. diff --git a/docs/user_guides/fs/feature_view/feature_logging.md b/docs/user_guides/fs/feature_view/feature_logging.md index 063e441076..c3e01e97a7 100644 --- a/docs/user_guides/fs/feature_view/feature_logging.md +++ b/docs/user_guides/fs/feature_view/feature_logging.md @@ -1,18 +1,18 @@ # User Guide: Feature and Prediction Logging with a Feature View -Feature logging is essential for debugging, monitoring, and auditing the data your models use. -This guide explains how to log features and predictions, and retrieve and manage these logs with feature view in Hopsworks. +Log features and predictions with a feature view, then retrieve them for debugging and monitoring. ## Feature and Prediction Logging After you have trained a model, you can log the features it uses and the predictions with the feature view used to create the training data for the model. -You can log either transformed or/and untransformed features values. +You can log transformed features, untransformed features, or both. ### Enabling Feature Logging To enable logging, set `logging_enabled=True` when creating the feature view. -Two feature groups will be created for storing transformed and untransformed features, but they are not visible in the UI. -The logged features will be written to the offline feature store every hour by scheduled materialization jobs which are created automatically. +One logging feature group stores transformed features, untransformed features, predictions, and logging metadata together. +Older feature views can retain separate transformed and untransformed logging groups. +The logged features are written to the offline feature store by a materialization job that is created automatically and runs on a schedule. ```python feature_view = fs.create_feature_view("name", query, logging_enabled=True) @@ -21,6 +21,57 @@ feature_view = fs.create_feature_view("name", query, logging_enabled=True) Alternatively, you can enable logging on an existing feature view by calling `feature_view.enable_logging()`. Also, calling `feature_view.log()` will implicitly enable logging if it has not already been enabled. +### Choosing the Transport + +A feature view logs through one of two transports, and the layout of its logging feature group follows from the choice. + +| Transport | Path of a logged row | Readable | +| --- | --- | --- | +| `realtime` (default) | The deployment posts Arrow batches to its inference logger, which produces them to Kafka; the online store receives them within seconds and the materialization job appends them to the offline store on its schedule | Online at once with `read_log(online=True)` for the group's time to live, offline after materialization | +| `job` | The deployment appends Arrow batches to a file buffer on its pod, rotates the buffer on size or age and uploads it to HopsFS; a scheduled commit job appends the uploaded chunks to an offline-only logging group | Offline after the commit job has run | + +A new feature view names its transport when logging is enabled: + +```python +feature_view = fs.create_feature_view( + "name", query, logging_enabled=True, logging_transport="job" +) +feature_view.feature_logging.transport # "job" +``` + +A feature view that does not log yet names it when logging is enabled, and the transport is read back from the view: + +```python +feature_view.enable_logging(transport="realtime") +feature_view.feature_logging.transport # "realtime" +``` + +The two cannot be combined on one feature view: enabling the other transport while the view logs is refused. +To move a view from one transport to the other, drop its log and recreate the logging group for the new transport with `feature_view.delete_log(transport="job")`. +Deployments take the transport from the view; a `DeploymentLoggingConfig` that names a different one is rejected. + +The `job` transport keeps no online copy, so `read_log(online=True)` is refused for such a view, and a deployment that stops uploads what its buffer holds and starts the commit job before the pod exits. +Run `deployment.commit_feature_logs()` or `feature_view.materialize_log()` to commit the uploaded chunks on demand, for example after a replica was killed. + +### Choosing the Materialization Interval { #choosing-the-materialization-interval } + +The materialization job runs every hour or once a day. +The platform default applies unless you choose one, at creation or later. + +```python +feature_view = fs.create_feature_view( + "name", query, logging_enabled=True, logging_materialization_interval="day" +) + +feature_view.enable_logging(materialization_interval="hour") + +feature_view.set_log_materialization_interval("day") +``` + +The interval only sets how often logs reach the offline store. +Run `feature_view.materialize_log()` to write them on demand between scheduled runs. +On the `job` transport the interval schedules the commit job instead. + ### Logging Features and Predictions You can log features and predictions by calling `feature_view.log`. @@ -198,9 +249,12 @@ feature_view.resume_logging() ## Materializing Logs -Besides the scheduled materialization job, you can materialize logs from Kafka to the offline store on demand. +Besides the scheduled materialization job, you can materialize logs to the offline store on demand. +On the `realtime` transport this reads the rows from Kafka. +On the `job` transport this runs the commit job over the chunks that deployments uploaded to HopsFS. This does not pause the scheduled job. -By default, it materializes both transformed and untransformed logs, optionally specifying whether to materialize transformed (transformed=True) or untransformed (transformed=False) logs. +Materialization writes all columns of the logging group. +The `transformed` selector applies only to older feature views with separate logging groups. ### Materialize Logs @@ -209,29 +263,38 @@ Materialize logs and optionally wait for the process to complete. ```python # Materialize logs and wait for completion materialization_result = feature_view.materialize_log(wait=True) -# Materialize only transformed log entries -feature_view.materialize_log(wait=True, transformed=True) ``` +## Monitoring Feature Logging + +A deployment that logs through the `realtime` transport reports what its inference logger is doing to Prometheus, and the deployment page shows it. +Open the deployment and look at the Feature logging card. +It shows four panels: rows logged per second by outcome, the time from a post to Kafka's acknowledgement, rows in flight, and posts per second by type and outcome. +The Full dashboard link opens the Feature Logging dashboard in Grafana, filtered to the same deployment, which adds in-flight bytes, rejected posts and totals over the selected range. + +Two of these answer most questions. +A non-zero rate of dropped or failed rows means the deployment logs faster than the inference logger can produce, or Kafka is refusing writes; the deployment logs name the reason. +Rejected posts mean the batches the predictor builds do not match the logging group's schema, which happens after the feature view changed without a redeploy. + +For a feature view on the `job` transport the card shows the same rows per second and buffered rows, the upload latency of a buffer segment to HopsFS, the bytes awaiting upload and the chunks uploaded per second; the predictor publishes these itself, and the Full dashboard adds commit job triggers and writer restarts. +The card is not shown for a view whose logging still runs through the row path of earlier releases. +Those logs are covered by the commit job's or the materialization job's own execution history instead. + ## Deleting Logs When log data is no longer needed, you might want to delete it to free up space and maintain data hygiene. This operation deletes the feature groups and recreates new ones. Scheduled materialization job and log timeline are reset as well. +Pass `transport="realtime"` or `transport="job"` to recreate the logging group for the other transport. ### Delete Logs -Remove all log entries (both transformed and untransformed logs), optionally specifying whether to delete transformed (transformed=True) or untransformed (transformed=False) logs. +Remove all log entries. +The `transformed` selector applies only to older feature views with separate logging groups. ```python # Delete all log entries feature_view.delete_log() - -# Delete only transformed log entries -feature_view.delete_log(transformed=True) ``` -## Summary - -Feature logging is a crucial part of maintaining and monitoring your machine learning workflows. -By following these examples, you can effectively log, retrieve, and delete logs, as well as manage the lifecycle of log materialization jobs, adding observability for your AI system and making it auditable. +Restart serving revisions after recreating a logging group so they load its new schema and destination. diff --git a/docs/user_guides/mlops/serving/api-protocol.md b/docs/user_guides/mlops/serving/api-protocol.md index a43df7a9c7..cb2aaaa6b5 100644 --- a/docs/user_guides/mlops/serving/api-protocol.md +++ b/docs/user_guides/mlops/serving/api-protocol.md @@ -1,10 +1,22 @@ -# How to Select the API protocol for a Deployment +# How to Select the API protocol for a Deployment { #api-protocol-guide } ## Introduction Hopsworks supports both REST and gRPC as API protocols for sending inference requests to model deployments. While REST API protocol is supported in all types of model deployments, gRPC is currently supported for **Python model deployments** only. +The protocol is chosen per deployment with `api_protocol`, in the creation form or in the Python API, and defaults to REST. +REST is what `curl`, the published OpenAPI document and any client that is not the Python library use. + +gRPC costs less per request under concurrency. +On a four-client benchmark it served 20 to 30 percent more requests per second and cut p99 latency by around 3 ms, and the gain grows with the batch size. +It is worth choosing when the Python library is the only client. +A deployment served by the [default predictor][deployment-schema] supports both protocols, because the library encodes the request and decodes the response at both ends. +On gRPC the rows travel as one KServe v2 tensor per schema field, and `deployment.predict()` returns the same dictionary it returns over REST. + +A deployment that runs your own predictor script has to stay on REST unless the script is written for gRPC. +Under gRPC the model server hands `predict()` KServe v2 tensors rather than rows, which a script written for REST cannot read. + ## Web UI ### Step 1: Create a new deployment @@ -50,6 +62,9 @@ You can select the API protocol to be enabled in your model deployment in the ad Therefore, only one of REST or gRPC API protocols can be enabled at the same time on the same model deployment. You cannot change the API protocol of existing deployments. + A gRPC deployment answers no HTTP requests, so `curl` cannot test it and the deployment page shows no curl example + and no OpenAPI reference for it. + Once you are done with the changes, click on `Create new deployment` at the bottom of the page to create the deployment for your model. ## Code diff --git a/docs/user_guides/mlops/serving/deployment-schema.md b/docs/user_guides/mlops/serving/deployment-schema.md index d28ef0167a..a85976ac11 100644 --- a/docs/user_guides/mlops/serving/deployment-schema.md +++ b/docs/user_guides/mlops/serving/deployment-schema.md @@ -12,7 +12,9 @@ A deployment schema lists the fields a client sends with each request, with thei It is inferred from the feature view the model was registered with: the serving keys, the features you pass with the request, the request parameters of on-demand transformations, and the extra columns of feature logging. It is published as a JSON Schema and an OpenAPI document, so clients in any language can validate requests before sending them. Every REST V1 request is validated in the pod before any predictor code runs, and rejected with a structured error when it does not match. -Enforcement covers the KServe REST V1 protocol only: a gRPC deployment is served without it, and the pod logs a warning at startup. +The wrapper that enforces it covers the KServe REST V1 protocol only. +A gRPC deployment served by the default predictor is still validated, by the predictor itself on the rows it decodes from the request tensors. +A gRPC deployment running your own script is not validated on either side, and the pod logs a warning at startup. The **default predictor** is the library class that serves such a deployment: it looks up and transforms the features by serving key, runs the model, and logs the request when the feature view has logging enabled. A [feature view can be deployed on its own][feature-view-deployment] with the same class and the same contract, returning the transformed feature vector instead of a prediction. @@ -67,9 +69,14 @@ The default predictor loads a single `.pkl`, `.pickle`, or `.joblib` file from t deployment.start(await_running=600) ``` -The default predictor is used when the model is a Python model registered with a feature view, no `script_file` or transformer is given, and the deployment uses KServe over REST. +The default predictor is used when the model is a Python model registered with a feature view, no `script_file` or transformer is given, and the deployment uses KServe. Pass `default_predictor=True` to force it, for instance for a scikit-learn model, or `default_predictor=False` to keep the plain model server. +Such a deployment serves either [API protocol][api-protocol-guide]. +It defaults to REST, which is what the `curl` example and the OpenAPI document below use. +Pass `api_protocol="GRPC"` to serve gRPC instead, which costs less per request under concurrency: the library owns both ends of the encoding, so the rows travel as one KServe v2 tensor per schema field and `deployment.predict()` returns the same dictionary it returns over REST. +A deployment serves one protocol, not both, so a gRPC deployment answers no HTTP and neither `curl` nor the OpenAPI document below reaches it. + At pod start the predictor checks that every input column of the model schema is served by the feature view, with a compatible type. A mismatch fails the deployment with the offending columns in `deployment.get_logs()`, instead of serving wrong predictions. @@ -155,7 +162,7 @@ A refinement keeps the inferred fields; adding or removing one is refused. ``` A custom predictor script deployed with `schema=` (or `passed_features=`) gets the same validation in the pod, before its `predict()` is called, for REST V1 requests. -The client validates REST requests only, so a gRPC client is not checked on either side. +A custom script asked to serve gRPC is checked on neither side, and has to read v2 tensors itself. ### Step 7: Republish after changing the feature view @@ -263,35 +270,61 @@ Declare the reserved extra logging columns on the feature view and the predictor Any other extra logging column becomes a request field that clients may send. -Logging is asynchronous: the request is answered immediately, the logging frame is built on a background thread of the predictor, and the rows are handed to the pod's inference-logger sidecar from there. -A logging failure never fails a request, and both buffers are bounded by `FEATURE_LOGGER_QUEUE_SIZE` rows (default 1000: rows waiting for the predictor's logging thread, and rows waiting in the sidecar logger); beyond it a request's rows are dropped and counted, so a slow logger cannot exhaust the pod's memory. -Both buffers count rows rather than requests, because one request carries a whole batch. -The predictor's own backlog admits one request whatever its size when it is empty, so a deployment whose batches are larger than the buffer logs instead of dropping every request; its peak is then that single batch, itself capped by the schema's batch limit. -There is no synchronous mode: a prediction is never delayed by its log write. - -## Custom predictor scripts +### Configuring feature logging per deployment { #deployment-schema-feature-logging-config } -Subclass the default predictor when the model needs another loader or the predictions need post-processing, and deploy with `default_predictor=True` so the schema is still inferred: +The predictor coalesces log rows into Arrow batches and hands them to the transport the feature view logs through: on `realtime` it posts them to the deployment's inference logger, which produces them to Kafka, and they reach the online store within seconds and the offline store on the materialization schedule; on `job` it appends them to a file buffer on the pod, which is uploaded to HopsFS and committed to the offline store by the view's commit job. +Both sides take their limits from platform variables that an administrator sets, and a deployment can override any of them with a `DeploymentLoggingConfig` (`hsml.deployment_logging_config`) passed to `deploy()`, `create_predictor()` or `feature_view.deploy()`, or set on the deployment before it starts. -=== "Python" - - ```python - from hsml.default_predictor import DefaultPredict +```python +from hsml.deployment_logging_config import DeploymentLoggingConfig +deployment = model.deploy( + feature_logging=DeploymentLoggingConfig( + batch_bytes=256 * 1024, # post once a quarter megabyte is waiting + batch_seconds=2, # or after two seconds under load + ) +) +deployment.start() +``` - class Predict(DefaultPredict): - def load_model(self, model_files_path): ... +A dict with the same field names is accepted wherever the object is. +Fields left unset keep the platform default. +The values are read when the pods start: edit `deployment.feature_logging`, call `deployment.save()`, and `deployment.restart()` a running deployment to apply them. - def model_predict(self, feature_vectors): - return self.model.predict_proba(feature_vectors[self.model_input_columns]) - ``` +```python +deployment.feature_logging.batch_seconds = 1 +deployment.save() +deployment.restart() +``` -The serving wrapper imports a model deployment's script itself, so the script needs no `__main__` block. -Only a [feature view deployment][feature-view-deployment] script, which may be started as a plain script, hands over to the wrapper. -Any predictor script, subclass or not, is protected by the serving wrapper when the deployment carries a schema: invalid rows and oversize batches are refused before `predict()` runs. -With a transformer, the transformer validates the request, whether or not it implements `preprocess()`, and the predictor trusts the transformer's output. -Each pod reads that role from its own revision, so a predictor created before a transformer was added keeps validating until it is replaced. -This needs an inference environment built from a Hopsworks 5.1 or later base image; an older image serves the deployment without checking. +| Field | Controls | Platform default | Variable | +| --- | --- | --- | --- | +| `transport` | The feature view's transport, `realtime` or `job`; a deployment cannot choose the other one, the field only documents or checks it | the view's | `serving_feature_logging_transport` (the default for new views) | +| `batch_bytes` | Coalesced bytes that force a post while the predictor has a backlog; an idle predictor posts at once | 1 MiB | `serving_feature_logger_batch_bytes` | +| `batch_seconds` | Longest time the predictor holds a partial batch before posting it | 5 | `serving_feature_logger_batch_seconds` | +| `batch_rows` | Most rows one post carries; the platform default is the inference logger's own limit, so a lower value only makes posts smaller | 512 | `serving_feature_logger_max_event_rows` | +| `queue_size` | Rows the predictor keeps queued for logging, including rows in flight, beyond which rows are dropped and counted; the queue holds the requests' rows as received, so wide rows hold more memory per row | 1000 | `serving_feature_logger_queue_size` | +| `max_event_bytes` | Largest single post; a group of requests larger than this goes out as several posts | 8 MiB | `serving_feature_logger_max_event_bytes` | +| `sidecar_cpu` | CPU request of the sidecar container, in cores | from the chart | inference logger values | +| `sidecar_memory_mb` | Memory request of the sidecar container, in MiB | from the chart | inference logger values | + +The `job` transport adds `flush_bytes` (1 MiB) and `flush_interval_seconds` (300), the size and age at which the pod's buffer segment is closed and uploaded, `max_buffer_bytes` (64 MiB), beyond which new rows are dropped while uploads fail, and `shutdown_seconds`, the budget a stopping pod has to upload its buffer and start the commit job; setting them on a `realtime` deployment is rejected. +Their variables are `serving_feature_logger_flush_bytes`, `serving_feature_logger_flush_interval_seconds`, `serving_feature_logger_max_buffer_bytes` and `serving_feature_logger_shutdown_seconds`. +A stopping pod is given a termination grace period of 30 seconds for the Knative drain plus `shutdown_seconds` plus 5, so a stop takes about that long to complete. +A pod that has uploaded 32 MiB since the last run asks the commit job to run ahead of its schedule, at most once every five minutes. +The object rejects non-positive values and inconsistent pairs: `batch_rows` cannot exceed `queue_size`, and `batch_bytes` cannot exceed `max_event_bytes`. +A post is closed as soon as the next request would take it past `batch_rows` or `max_event_bytes`, so a backlog is posted in receiver-sized pieces. + +Under load a lower `batch_bytes` or `batch_seconds` shortens the time a row waits in the predictor at the cost of more posts; when the deployment is idle every row is posted as soon as it is built, whatever the values. +Stopping a deployment posts whatever the predictor still holds before the pod exits; on the `job` transport it uploads the buffer and starts the commit job, and `deployment.commit_feature_logs()` runs that job on demand. + +Logging is asynchronous and a logging failure does not fail prediction. +How often the rows reach the offline store is a property of the feature view, not the deployment; see [Choosing the Materialization Interval][choosing-the-materialization-interval]. + +A deployment that logs features serves on one worker process by default, whatever its CPU limit. +The rows are safe with several: each worker buffers separately and the buffer directory lock arbitrates who adopts a dead worker's segments. +The metrics are not: they are this process's counters, read when Prometheus scrapes, so with several workers behind one port a scrape reaches one of them and the Feature logging card reports a fraction of the rows. +Set `KSERVE_WORKERS` in `env_vars=` to serve on more than one anyway, and read the card as a sample rather than a total. ## Deployments without lookups { #deployment-schema-no-lookup } @@ -353,9 +386,12 @@ Log rows contain feature values and are governed by the logging feature group's | `SERVING_TRAINING_DATASET_VERSION` | the client, feature view deployments | the pinned training dataset | | `SERVING_SCHEMA_ENFORCER` | the client | `predictor` or `transformer`: the component of the revision that validates requests | | `SERVING_MAX_BATCH_ROWS` | you, through `env_vars=` | rows accepted per request, default 512; recorded in the schema at publication | -| `FEATURE_LOGGER_QUEUE_SIZE` | you, through `env_vars=` | rows the predictor's logging thread and the async logger each buffer before dropping, default 1000; a value that is not a positive integer is ignored | +| `SERVING_PREDICTOR_ASYNC_LOOKUP` | you, through `env_vars=` | `false` returns the default predictor to the blocking online store lookup | +| `KSERVE_WORKERS` | you, through `env_vars=` | uvicorn worker processes; the default is one per whole core, capped at 4, and one on a deployment that logs features | +| `HOPSWORKS_FEATURE_LOGGING_TRANSPORT` | the backend | the transport the view's logging group uses, set only on a deployment that logs | -The `SERVING_*` names are reserved and refused in `env_vars=`, except `SERVING_MAX_BATCH_ROWS`. +The `HOPSWORKS_*` names are reserved and refused in `env_vars=`, as are `SERVING_SCHEMA_ID`, `SERVING_FEATURE_VIEW_NAME`, `SERVING_FEATURE_VIEW_VERSION`, `SERVING_TRAINING_DATASET_VERSION` and `SERVING_SCHEMA_ENFORCER`. +The logging limits are set through `DeploymentLoggingConfig` rather than through `env_vars=`: `FEATURE_LOGGER_QUEUE_SIZE`, `FEATURE_LOGGER_BATCH_BYTES` and `FEATURE_LOGGER_BATCH_SECONDS` are reserved, so a value set there is refused. ## API Reference diff --git a/docs/user_guides/mlops/serving/predictor.md b/docs/user_guides/mlops/serving/predictor.md index 82a82d3f9d..800f0a86fa 100644 --- a/docs/user_guides/mlops/serving/predictor.md +++ b/docs/user_guides/mlops/serving/predictor.md @@ -197,6 +197,25 @@ The serving wrapper imports a model deployment's script itself, so the script ne Only a [feature view deployment][feature-view-deployment] script, which may be started as a plain script, hands over to the wrapper. See the [Deployment Schema Guide][deployment-schema] for the request contract, the error codes, and the feature logging guarantees. +!!! note "The default predictor's `predict` is a coroutine" + `DefaultPredict.predict` is `async def`, and it awaits the online store lookup rather than blocking on it. + The lookup is a round trip, so on a blocking predictor it held the server's event loop and stopped every other request in the deployment for its duration: measured, that capped a deployment at 218 requests per second where the same deployment with nothing to look up reached 310. + Awaiting it raised throughput by 23.6 percent and cut p99 latency by 72 percent. + + A subclass overriding `model_predict` or `load_model` is unaffected, since neither is a coroutine. + A subclass overriding `predict` itself must declare it `async def` and `await super().predict(...)`. + + To drive the predictor from a script or a notebook, where there is no event loop to hold up, call `predict_blocking` instead. + It serves the same request through the same body and returns the same value; called from inside a running loop it refuses rather than deadlocks. + + ```python + predictor = Predict() + result = predictor.predict_blocking([{"cc_num": 1234}]) + ``` + + Set `SERVING_PREDICTOR_ASYNC_LOOKUP=false` on the deployment to go back to the blocking lookup. + That is worth doing only when the deployment reads the online store through the REST client, where there is nothing to overlap. + To serve the model with your own code instead, implement a predictor script (Steps 2.1 and 2.2). ### Step 2.1 (Optional): Implement a predictor script diff --git a/mkdocs.yml b/mkdocs.yml index 68e3f7aec6..29df62be80 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -82,6 +82,7 @@ nav: - user_guides/fs/feature_group/index.md - Create: user_guides/fs/feature_group/create.md - Partitioning and Clustering: user_guides/fs/feature_group/partitioning.md + - Delta Maintenance: user_guides/fs/feature_group/delta_maintenance.md - Create External: user_guides/fs/feature_group/create_external.md - Ingest Data with dltHub: user_guides/fs/feature_group/ingest_with_dlthub.md - Create Spine: user_guides/fs/feature_group/create_spine.md