Run an AI Model with Confluent Cloud for Apache Flink

Confluent Cloud for Apache Flink® supports AI model inference and enables using models as resources in Flink SQL, just like tables and functions. You can use a SQL statement to create a model resource and pass it on for inference in streaming queries. The SQL interface is available in Cloud Console and the Flink SQL shell.

The CREATE MODEL statement registers an AI or ML model in your Flink environment for real-time prediction and inference. You need to create a model before you can use it in your queries.

The following examples get you started with AI model inference.

Prerequisites

  • Access to Confluent Cloud.
  • Access to a Flink compute pool.
  • Sufficient permissions to create models. For more information, see RBAC for model inference.

Create an AI model

You need an AI model on one of the cloud providers. The following steps show how to start creating models.

  1. In your web browser, navigate to the AWS Bedrock site.

  2. Pick a Foundation Model. You may need to request access and accept a EULA, if your account hasn’t before.

  3. At the bottom of the page, get the modelId of the model from the API request example.

    The model endpoint resembles https://bedrock-runtime.<REGION>.amazonaws.com/model/<MODEL_ID>/invoke.

  4. Bedrock doesn’t use API keys. Instead, you must get your AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY. If you’re using temporary credentials, you need an AWS_SESSION_TOKEN, which you can get from the AWS access portal. If you use temporary credentials, the query may not work after the credentials expire.

  5. Set the INPUT_FORMAT associated with your chosen model from the list above. If not set, Confluent Cloud for Apache Flink usually chooses the correct one automatically based on the name of the model.


Create a connection resource

A connection resource enables you to connect to model providers in a way that protects your secrets, so Flink statements can make calls to these services securely.

Note

Connection resources are an Open Preview feature in Confluent Cloud.

A Preview feature is a Confluent Cloud component that is being introduced to gain early feedback from developers. Preview features can be used for evaluation and non-production testing purposes or to provide feedback to Confluent. The warranty, SLA, and Support Services provisions of your agreement with Confluent do not apply to Preview features. Confluent may discontinue providing preview releases of the Preview features at any time in Confluent’s’ sole discretion.

Run the CREATE CONNECTION statement to create a new connection resource.

The following examples show how to use connection resources in the CREATE MODEL statements that register models with Confluent Cloud for Apache Flink.

  • Connections must be created in the same cloud region as the corresponding models.

For details on reusable connections, see Reuse Confluent Cloud Connections With External Services and Manage Connections with External Services in Confluent Cloud.

Text embedding with AWS Bedrock and Azure OpenAI

This example uses the us-central1 region on Google Cloud, but for best performance, you should use a regional endpoint near the location where the Flink statement will run.

The input and output formats for the embedding model are detected automatically.

  • In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create the table for input text.

    CREATE TABLE `text_input` (
  `id` STRING,
  `input` STRING
);

INSERT INTO text_input VALUES
  ('1', 'alien'),
  ('2', 'golden finger'),
  ('3', 'license to kill'),
  ('4', 'aliens');

AWS Bedrock

For simplicity, this example uses the AWS Bedrock “amazon.titan-embed-text-v1” model.

  1. Run the following statement to create a connection resource named bedrock-cli-connection that uses your AWS credentials.

    CREATE CONNECTION bedrock-cli-connection
WITH (
  'type' = 'bedrock',
  'endpoint' = 'https://bedrock-runtime.<REGION>.amazonaws.com/model/<MODEL_ID>/invoke',
  'aws_access_key_id' = '<your-aws-access-key-id>',
  'aws_secret_access_key' = '<your-aws-secret-access-key>',
  'aws_session_token' = '<your-aws-session-token>'
);

  2. Run the following statement to register your model with Confluent Cloud for Apache Flink.

    CREATE MODEL bedrock_embed
INPUT (text STRING)
OUTPUT (response ARRAY<FLOAT>)
WITH (
  'bedrock.connection'='bedrock-cli-connection',
  'bedrock.input_format'='AMAZON-TITAN-EMBED',
  'provider'='bedrock',
  'task'='embedding'
);

  3. Run the following statement to invoke your AI model.

    SELECT * from text_input, LATERAL TABLE(ML_PREDICT('bedrock_embed', input));

Azure OpenAI

  1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create a connection resource named azureopenai-cli-connection that uses your Azure API key.

    CREATE CONNECTION azureopenai-cli-connection
WITH (
  'type' = 'azureopenai',
  'endpoint' = 'https://<your-resource-name>.openai.azure.com/openai/deployments/<your-deployment-name>/embeddings?api-version=2024-06-01',
  'api-key' = '<your-azure-api-key>'
);

  2. Run the following statement to register your model with Confluent Cloud for Apache Flink.

    CREATE MODEL azure_embed
INPUT (text STRING)
OUTPUT (response ARRAY<FLOAT>)
WITH (
  'azureopenai.connection'='azureopenai-cli-connection',
  'provider'='azureopenai',
  'task'='embedding'
);

  3. Run the following statement to invoke your AI model.

    SELECT * from text_input, LATERAL TABLE(ML_PREDICT('azure_embed', input));

Google AI

  1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create a connection resource named googleai-cli-connection that uses your Google Cloud API key.

    CREATE CONNECTION googleai-cli-connection
WITH (
  'type' = 'googleai',
  'endpoint' = 'https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent',
  'api-key' = '<your-gcp-api-key>'
);

  2. Run the following statement to register your model with Confluent Cloud for Apache Flink.

    CREATE MODEL google_text_cli
INPUT (`text` VARCHAR(2147483647))
OUTPUT (`output` VARCHAR(2147483647))
WITH (
  'googleai.connection' = 'googleai-cli-connection',
  'googleai.system_prompt' = 'translate text to chinese',
  'provider' = 'googleai',
  'task' = 'text_generation'
);

  3. Run the following statement to invoke your AI model.

    SELECT * FROM google_input, LATERAL TABLE(ML_PREDICT('google_text_cli', text));

Sentiment analysis with OpenAI LLM

  1. In a Confluent Cloud Console workspace or the Flink SQL shell, run the following statement to create the table for input text.

    CREATE TABLE text_stream (
  id BIGINT, text STRING
);

INSERT INTO text_stream SELECT 1 id, 'The mitochondria are the powerhouse of the cell' text;
INSERT INTO text_stream SELECT 2 id, 'Happy Birthday! You are great!' text;
INSERT INTO text_stream SELECT 3 id, 'Today was bad day in the stock market.' text;

  2. Run the following statement to create a connection resource named openai-connection that uses your OpenAI API key.

    CREATE CONNECTION openai-connection
WITH (
  'type' = 'openai',
  'endpoint' = 'https://api.openai.com/v1/chat/completions',
  'api-key' = '<your-api-key>'
);

  3. Run the following statement to create the OpenAI model with a system prompt for sentiment analysis.

    CREATE MODEL sentimentmodel
INPUT(text STRING)
OUTPUT(sentiment STRING)
COMMENT 'sentiment analysis model'
WITH (
  'provider' = 'openai',
  'task' = 'classification',
  'openai.connection' = 'openai-connection',
  'openai.model_version' = 'gpt-3.5-turbo',
  'openai.system_prompt' = 'Analyze the sentiment of the text and return only POSITIVE, NEGATIVE, or NEUTRAL.'
);

  4. Run the inference statement on the table and model.

    SELECT id, text, sentiment FROM text_stream, LATERAL TABLE(ML_PREDICT('sentimentmodel', text));

Related content

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.