Model deployment with Kafka

Verta supports near real-time/asynchronous models with Kafka integration, allowing Verta endpoints to be configured to read data from a Kafka topic, perform predictions, and write the output to another Kafka topic. In case of any error scenarios, Verta sends the errors to an error topic for further analysis or handling.

In this guide, we will walk through the steps to take any model and deploy it in a scalable infrastructure using Verta and Kafka.

Context

• A Kafka message is an arbitrary key and value.

• Verta endpoint is a containerized microservice serving a model that allows you to make real-time predictions. Learn more about Verta Model Deployment.

• The inputs to a Verta endpoint need to be JSON.

A Verta endpoint is a containerized microservice serving a model that allows you to make real-time predictions. Kafka messages are arbitrary keys and values. If the values of messages in a Kafka topic are valid JSON, conforming to the expected structure of a deployed model, that Kafka topic can be used as the input topic for a Verta endpoint.

Setup

Pre-requisites:

To get started, you need the following setup;

• A packaged model ready to be deployed as an endpoint in Verta

• A Verta installation configured to talk to Kafka brokers. Please refer to Kafka integration system admin guide to learn how to setup the integration.

• Kafka topics to be used as the model input source, and the destination for model outputs and errors. Use client function get_kafka_topics() to fetch a list of available Kafka topics for the current configuration.

Note: Every endpoint should be uniquely associated with an input, output and error topic. Topics should not be re-used across endpoints.

Steps for deployment with Kafka:

Using Web UI

• During the deployment process, you can enable Kafka integration and provide the necessary Kafka topics. These topics include the input topic, output topic, and error topic, which should be set up in your Kafka cluster. You will have the option to select the appropriate topics from a list of available topics in your Kafka cluster.

• Once the endpoint is live and active, you will be able to track the status of Kafka endpoint configuration from the endpoint summary and settings tabs.

• To enable or update Kafka configuration for an existing live endpoint or during one-click deployment, go to the endpoint's settings tab.

Using Client Example to add Kafka setting when deploying an endpoint:

from verta.endpoint import KafkaSettings
from verta import Client

client = Client()  # Initialize from environment variables

# List available Kafka topics.
kafka_topics = client.get_kafka_topics()
print("\nAvailable Kafka Topics:")
print(kafka_topics)

kafka_settings = KafkaSettings(
    input_topic="my_input_topic",
    output_topic="my_output_topic",
    error_topic="my_error_topic",
)

endpoint = client.create_endpoint(path="/some-path", kafka_settings=kafka_settings)
print(endpoint.kafka_settings)

Example to update Kafka setting for a deployed endpoint:

from verta.endpoint.update import DirectUpdateStrategy
from verta.endpoint import KafkaSettings

new_kafka_settings = KafkaSettings(
    input_topic="new_input_topic",
    output_topic="new_output_topic",
    error_topic="new_error_topic",
)

endpoint.update(model_version, DirectUpdateStrategy(), kafka_settings=new_kafka_settings)

Example to update Kafka setting re-using an existing build for a deployed endpoint:

from verta.endpoint.update import DirectUpdateStrategy
from verta.endpoint import KafkaSettings

new_kafka_settings = KafkaSettings(
    input_topic="new_input_topic",
    output_topic="new_output_topic",
    error_topic="new_error_topic",
)

endpoint.update(
    endpoint.get_current_build(),
    kafka_settings=new_kafka_settings,
)

How does it work?

For an endpoint configured to read from and write to Kafka this is how the Kafka messages will be published and subscribed to;

• The Verta consumer will subscribe to the input topic.

• Verta will only process messages sent to Kafka after this subscription starts.

• The message key will be passed through unchanged to output and error topics. Metadata stored in the message key may help in tracing, auditing, and debugging model behavior. For example, the message key might be a unique identifier like UUID, or a JSON object which contains metadata.

• The input topic message value is sent as the HTTP request body. The message value should be JSON conforming to the expected model input schema.

• For successful responses, the response body will be used as the value of a message written to the output topic.

• For unsuccessful responses, e.g. any HTTP error message in the 400 range, the response body will be used as the value of a message written to the error topic.