# Set Up AI Functions in Confluent Manager for Apache Flink #### IMPORTANT `AI_FORECAST` and `AI_DETECT_ANOMALIES` are a preview feature. Interfaces might change in future releases. Confluent Manager for Apache Flink® (CMF) can run two Flink SQL functions that call time-series foundation models: `AI_FORECAST` forecasts the next values of a series, and `AI_DETECT_ANOMALIES` flags values that fall outside the range a model predicts. On Confluent Platform for Apache Flink, the models run in model servers that you deploy in your Kubernetes cluster, and the functions send a statement’s values to those servers. This page is for the administrator who sets up the functions. To call them, see [Forecast with AI_FORECAST](forecast.md#af-cp-ai-forecast) and [Detect anomalies with AI_DETECT_ANOMALIES](detect-anomalies.md#af-cp-ai-detect-anomalies). ## How the functions reach the models Each model runs in its own model server, a Kubernetes Deployment with a Service in front of it. The functions ship in a JAR that you upload to an environment and register with `CREATE FUNCTION`. You then set the URL of each model server in the environment’s statement defaults. CMF passes the URLs to the job of every statement in the environment, so a call names only the model, and statement authors can reach only the servers that you set. The model server image serves the following models: | Model | Name | Description | |-------------|-------------|----------------------------------------------------------------------------------------------------------------| | TTM-R3 | `ttm` | IBM Granite TinyTimeMixer, a small general-purpose model. The
functions use it when a call names no model. | | FlowState | `flowstate` | IBM Research FlowState. | | PatchTST-FM | `patchtst` | IBM Granite PatchTST-FM. Its model download is about 1 GB. | You set up only the models that your statements use. ## Requirements - CMF must have the [environment catalog](../../../configure/environment-catalog.md#cmf-environment-catalog-enable) and [artifact management](../../../configure/artifacts.md#cmf-artifacts-enable) enabled. - CMF must have compile workers, because only compile workers load function JARs. - Statements that call the functions must run on a compute pool that uses Flink 2.3. - The model server image is for amd64 and arm64 nodes. On amd64, the CPU must support the x86-64-v3 instruction set, which includes AVX2. - At start-up, the model server downloads its model from Hugging Face. If the pods cannot reach `huggingface.co`, provide the model files as described in [Run without internet access](). ## Deploy the model servers Deploy one model server for each model. The following manifest deploys the server for TTM-R3. The `MODEL` environment variable selects the model, and takes the names from the table above. ```yaml apiVersion: apps/v1 kind: Deployment metadata: name: model-server-ttm spec: replicas: 1 selector: matchLabels: app: model-server-ttm template: metadata: labels: app: model-server-ttm spec: containers: - name: model-server image: confluentinc/cp-flink-models:1.0.0 env: - name: MODEL value: ttm ports: - containerPort: 8080 readinessProbe: httpGet: path: /readyz port: 8080 periodSeconds: 5 failureThreshold: 120 resources: requests: cpu: "1" memory: 1Gi limits: cpu: "4" memory: 3Gi --- apiVersion: v1 kind: Service metadata: name: model-server-ttm spec: selector: app: model-server-ttm ports: - port: 8080 targetPort: 8080 ``` Apply it in a namespace that the Flink pods can reach, for example the environment’s namespace, `flink` in this example: ```bash kubectl apply -n flink -f model-server-ttm.yaml ``` The server answers requests at `/v1/forecast` on port 8080, so the URL of this server is `http://model-server-ttm.flink.svc.cluster.local:8080/v1/forecast`. Both functions use this URL. The pod becomes ready when `GET /readyz` returns 200, which happens after the server has downloaded and loaded its model. If the server cannot load the model, the container exits. A server for one model refuses requests for another model with HTTP 400, so a URL that points to the wrong server fails the statement instead of returning another model’s forecast. Give the container a CPU limit. The server runs one thread for each CPU of the limit. Without a limit, it runs one thread for each CPU of the node, and requests that arrive at the same time slow each other down. The server keeps no state between requests, so to serve more statements, add replicas. ### Keep the model download The server downloads the model again each time its container starts. To keep the download across restarts, mount a writable volume, such as a PersistentVolumeClaim, at `/home/appuser/.cache/huggingface`. ### Run without internet access To run the server without access to Hugging Face, download the model’s `config.json` and `model.safetensors` files at the revision the server expects, and mount the directory that holds them at `/mnt/models`. The server then loads the files from there and does not download anything. If a file is missing or is for another model, the container exits. For example, download the TTM-R3 files with the Hugging Face CLI: ```bash hf download ibm-granite/granite-timeseries-ttm-r3 config.json model.safetensors \ --revision 7b070728ff280ee2bb682bd8dfbe71da21ca2c7c --local-dir ttm ``` The following table lists the repository and revision for each model: | Name | Hugging Face repository | Revision | |-------------|-------------------------------------------------|--------------------------------------------| | `ttm` | `ibm-granite/granite-timeseries-ttm-r3` | `7b070728ff280ee2bb682bd8dfbe71da21ca2c7c` | | `flowstate` | `ibm-research/flowstate` | `940580f969036cfca514cdff725f7b02c03e1bcf` | | `patchtst` | `ibm-granite/granite-timeseries-patchtst-fm-r1` | `151f9c6d576281b95c2ff784d0863bd3f12c80f1` | ## Register the functions The functions ship in one JAR, `io.confluent.flink:cp-flink-ai-detect-anomalies:1.0.0`, in the Confluent Maven repository at `https://packages.confluent.io/maven/`. It holds both functions. Register them in each environment whose statements call them. 1. Download the JAR: ```bash curl -O https://packages.confluent.io/maven/io/confluent/flink/cp-flink-ai-detect-anomalies/1.0.0/cp-flink-ai-detect-anomalies-1.0.0.jar ``` 2. Upload it to the environment as an artifact. Put the artifact metadata in a file, `artifact.json`: ```json { "apiVersion": "cmf.confluent.io/v1", "kind": "Artifact", "metadata": { "name": "ai-functions.jar" }, "spec": {} } ``` Then upload the JAR to the `prod` environment: ```bash curl -X POST http://cmf:8080/cmf/api/v1/environments/prod/artifacts \ -F 'artifact=@artifact.json;type=application/json' \ -F 'file=@cp-flink-ai-detect-anomalies-1.0.0.jar' ``` For details, see [Upload an artifact](../../../configure/artifacts.md#cmf-artifacts-upload). 3. Register each function in the environment catalog with `CREATE FUNCTION`: ```bash confluent flink statement create create-ai-forecast \ --url http://cmf:8080 \ --environment prod \ --compute-pool pool \ --catalog _env_prod \ --database default \ --sql "CREATE FUNCTION AI_FORECAST AS 'io.confluent.cpflink.udf.ai.AIForecastFunction' USING JAR 'cmf://prod/ai-functions.jar';" confluent flink statement create create-ai-detect-anomalies \ --url http://cmf:8080 \ --environment prod \ --compute-pool pool \ --catalog _env_prod \ --database default \ --sql "CREATE FUNCTION AI_DETECT_ANOMALIES AS 'io.confluent.cpflink.udf.ai.AIDetectAnomaliesFunction' USING JAR 'cmf://prod/ai-functions.jar';" ``` You can choose other names, but the examples in this documentation use these. For other options of `CREATE FUNCTION`, see [User-Defined Functions in Confluent Manager for Apache Flink](../user-defined-functions.md#cmf-sql-udf). ## Set the model server URLs Set the URL of each model server in the environment’s statement defaults, as the Flink configuration key `ai.models..endpoint`, where `` is the model’s name from the table above. Set the keys in both blocks: `interactive` applies to `SELECT` statements, and `detached` applies to `INSERT INTO` statements and statement sets. For example, put the following statement defaults in a file, `env-prod-statement-defaults.json`: ```json { "interactive": { "flinkConfiguration": { "ai.models.ttm.endpoint": "http://model-server-ttm.flink.svc.cluster.local:8080/v1/forecast" } }, "detached": { "flinkConfiguration": { "ai.models.ttm.endpoint": "http://model-server-ttm.flink.svc.cluster.local:8080/v1/forecast" } } } ``` Then update the environment with it. An update replaces all of the environment’s defaults, so pass the files for its other defaults as well. For details, see [Configure Environments in Confluent Manager for Apache Flink](../../../configure/environments.md#cmf-environments). ```bash confluent flink environment update prod \ --url http://cmf:8080 \ --statement-defaults env-prod-statement-defaults.json ``` CMF rejects the update in the following cases: - The value is not an `http` or `https` URL with a host. - The URL contains a user name, a password, a query string, or a fragment. The URL appears in the job’s configuration and in the Flink web UI, so it must not carry credentials. - The statements of the environment do not compile for Flink 2.x, for example because CMF has no compiler for Flink 2.3. - The environment sets its first URL while statements that can still be resumed set `pipeline.global-job-parameters` themselves. Delete these statements first. After an environment sets a URL, CMF passes the URLs to the job of every statement in the environment that runs on an Flink 2.x compute pool, as job parameters. Statements in the environment can no longer set `pipeline.global-job-parameters` or keys that start with `ai.models.`, and CMF rejects them if they do. A call to a function then uses the URL that the environment sets for the model the call names. A call that names a model with no URL, or that names its own URL with the `endpoint` key, fails the statement with a message that says so. ### Change a URL When you change a URL, new statements use the new URL. A running statement keeps the URL it started with until CMF deploys it again, for example when you stop and resume it. Whether it keeps its state depends on its upgrade mode: with `stateless`, it starts again without state. For details, see [Manage Statements in Confluent Manager for Apache Flink](../manage-statements.md#flink-sql-statements-manage). Before you move a model server, keep the old URL answering until you have resumed the statements that use it. ## Check the setup Run a query that calls a function on each model that you set up, from a statement whose current catalog is `_env_prod` and current database is `default`. For example queries, see [Example](forecast.md#af-cp-ai-forecast-example) and [Example](detect-anomalies.md#af-cp-ai-detect-anomalies-example). ## Security and data handling - The functions send the values of a series, up to the last 512 values by default, to the model server, without their timestamps. They send nothing to Confluent. - The model server downloads model files from Hugging Face at start-up. It sends no data from your statements there. - The model server has no authentication and serves plain HTTP. Any pod that can reach its Service can call it. Restrict access to the Flink pods with a Kubernetes NetworkPolicy. - In an environment that sets no model server URL, each call names its server in the `endpoint` key of its configuration, and the job sends the values to whatever URL the statement author chose. Set the URLs in every environment where you register the functions. - To control who can upload artifacts, register functions, and change an environment’s statement defaults, see [Configure Access Control for Confluent Manager for Apache Flink](../../../configure/access-control.md#cmf-access-control). ## Troubleshoot A function checks its configuration when the first row reaches it. If it refuses the configuration, or if the model server refuses the request with an HTTP 4xx status other than 408, 425, or 429, the statement fails and does not restart. Fix the statement and create it again. The statement’s exceptions show the reason. The following messages are the most common: `model 'flowstate' has no model server URL in this environment` : The environment sets no `ai.models.flowstate.endpoint`. Set it, or call a model that has a URL. The message lists the models that have one. `the config must not name an endpoint` : The environment sets the URLs, so the call’s configuration must not contain `endpoint`. Remove it. `Infer failed: HTTP 400` : The model server refused the request, for example because the call asks for more than 96 steps, or because the URL points to the server of another model. If the model server does not answer within 10 seconds, cannot be reached, or answers with HTTP 408, 425, 429, or a 5xx status, the job restarts and calls the server again. Check that the model server pods are ready. ## Related content - [Forecast with AI_FORECAST](forecast.md#af-cp-ai-forecast) - [Detect anomalies with AI_DETECT_ANOMALIES](detect-anomalies.md#af-cp-ai-detect-anomalies) - [User-Defined Functions in Confluent Manager for Apache Flink](../user-defined-functions.md#cmf-sql-udf) - [Manage Artifacts in Confluent Manager for Apache Flink](../../../configure/artifacts.md#cmf-artifacts) - [Configure Environments in Confluent Manager for Apache Flink](../../../configure/environments.md#cmf-environments)