Run Model Inference in Confluent Manager for Apache Flink

With CREATE MODEL, you register a remote AI model in the environment catalog, and a statement calls it for each row with ML_PREDICT. Confluent Manager for Apache Flink® (CMF) supports the openai provider, which works with any service that has an OpenAI-compatible API.

To use models, you need:

The environment catalog, enabled. See Enable the environment catalog.
A compute pool that runs Flink 2.x. CMF rejects model statements on Flink 1.x compute pools.
The model’s endpoint in the list of allowed endpoints. See Allow model endpoints.

Create a model

Create the model in the environment catalog. The following example creates the model sentiment in the environment env-1:

confluent flink statement create create-sentiment-model \
  --url http://cmf:8080 \
  --environment env-1 \
  --compute-pool pool \
  --catalog _env_env-1 \
  --database default \
  --sql "CREATE MODEL sentiment INPUT (review STRING) OUTPUT (label STRING) WITH ('provider' = 'openai', 'endpoint' = 'https://api.openai.com/v1/chat/completions', 'model' = 'gpt-4o-mini', 'api-key' = '<api-key>', 'system-prompt' = 'Classify the review as positive or negative. Output only the label.');"

Option	Description
`provider`	Required. Must be `openai`.
`model`	Required. The name of the model on the model server.
`endpoint`	The URL of the model server. The path must end with `chat/completions` or `embeddings`. Required unless `credentials-secret-id` is set.
`api-key`	The key for the model server. Required unless `credentials-secret-id` is set.
`credentials-secret-id`	The name of a secret mapping that holds the endpoint and the key. See Keep the key in a secret.
`system-prompt`	The instructions that the model gets with each row.

CMF masks the value of api-key in the statement text and in the output of SHOW CREATE MODEL.

Keep the key in a secret

To keep the key out of the SQL text, store the endpoint and the key in a secret, and map the secret into the environment. The secret’s data must contain exactly the keys endpoint and api-key:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Secret",
  "metadata": {
    "name": "openai-secret"
  },
  "spec": {
    "data": {
      "endpoint": "https://api.openai.com/v1/chat/completions",
      "api-key": "<api-key>"
    }
  }
}

Create a secret mapping named openai-credentials in env-1 that points to openai-secret. For how to create the secret and the mapping, see Secrets and secret mappings. Then set credentials-secret-id to the name of the mapping:

CREATE MODEL sentiment INPUT (review STRING) OUTPUT (label STRING) WITH (
  'provider' = 'openai',
  'credentials-secret-id' = 'openai-credentials',
  'model' = 'gpt-4o-mini'
);

A model with credentials-secret-id can’t also set api-key. If it sets endpoint, the value must be the endpoint in the secret.

Call a model

Call the model with ML_PREDICT. Pass the input table, the model, and the columns to send to the model:

SELECT * FROM ML_PREDICT(TABLE reviews, MODEL sentiment, DESCRIPTOR(review));

The result has the columns of the input table and the OUTPUT columns of the model. Use the catalog and database to name a table or model outside the current database, for example TABLE `kafka-cat`.`kafka-db`.`reviews`.

ML_PREDICT sends the values of the DESCRIPTOR columns to the model’s endpoint.

Manage models

Run these statements in the environment catalog:

SHOW MODELS;
SHOW CREATE MODEL sentiment;
ALTER MODEL sentiment SET ('model' = 'gpt-4.1-mini');
ALTER MODEL sentiment RENAME TO review_sentiment;
DROP MODEL review_sentiment;

ALTER MODEL ... SET changes the options you name and keeps the others. A new statement that calls a dropped model fails. A statement that is already running keeps calling it.

Allow model endpoints

An administrator lists the endpoints that models may call in the CMF Helm values. The list is empty by default, so no model can be created until you add an endpoint:

cmf:
  sql:
    models:
      allowedEndpoints:
        - https://api.openai.com/v1

Each entry is an http or https URL with an optional path. An endpoint is allowed when its scheme, host, and port match an entry and its path is the entry’s path or below it. CMF doesn’t start if an entry isn’t a valid URL, or has user info, a query, or a fragment.

CMF checks the endpoint at CREATE MODEL, at ALTER MODEL ... SET, and each time a statement that calls the model is compiled. This includes an endpoint from a secret. A statement that is already running keeps calling its endpoint.

Limitations

The only provider is openai.
Models are stored only in the environment catalog, not in Kafka catalogs.
Model statements run only on compute pools with Flink 2.x.