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 and Detect anomalies with 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 and artifact management 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.

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:

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:

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.

Download the JAR:

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

Upload it to the environment as an artifact. Put the artifact metadata in a file, artifact.json:

{
  "apiVersion": "cmf.confluent.io/v1",
  "kind": "Artifact",
  "metadata": {
    "name": "ai-functions.jar"
  },
  "spec": {}
}

Then upload the JAR to the prod environment:

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.

Register each function in the environment catalog with CREATE FUNCTION:

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.

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.<name>.endpoint, where <name> 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:

{
  "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.

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.

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 and 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.

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.