# CREATE MODEL Statement in Confluent Cloud for Apache Flink Confluent Cloud for Apache Flink® enables real-time inference and prediction with AI and ML models. The Flink SQL interface is available in Cloud Console and the Flink SQL shell. Get started using AI models with these guides: - [Run a managed AI model](../../../ai/managed-model.md#flink-sql-managed-ai-model) - [Run a remote AI Model](../../../ai/ai-model-inference.md#flink-sql-ai-model) The following providers are supported: - AWS Bedrock - AWS Sagemaker - Azure Machine Learning (Azure ML) - Azure OpenAI - Google AI - OpenAI - Vertex AI Also, you can use fully managed models in Confluent Cloud by specifying the [confluent](#flink-sql-create-model-with-confluent) provider. ## Syntax ```sql CREATE MODEL [IF NOT EXISTS] [[catalogname].[database_name]].model_name [INPUT (input_column_list)] [OUTPUT (output_column_list)] [COMMENT model_comment] WITH(model_option_list) ``` ## Description Create a new AI model. If a model with the same name exists already, a new [version](#flink-sql-create-model-input-model-versioning) of the model is created. For more information, see [version](#flink-sql-create-model-input-model-versioning). If the IF NOT EXISTS option is specified and a model with the same name exists already, the statement is ignored. To view the currently registered models, use the [SHOW MODELS](show.md#flink-sql-show-models) statement. To view the [WITH options](#flink-sql-create-model-with-options) that were used to create the model, run the [SHOW CREATE MODEL](show.md#flink-sql-show-create-model) statement. To view the versions, inputs, and outputs of the model, run the [Models](describe.md#flink-sql-describe-model) statement. To change the name or options of an existing model, use the [ALTER MODEL](alter-model.md#flink-sql-alter-model) statement. To delete a model from the current environment, use the [DROP MODEL](drop-model.md#flink-sql-drop-model) statement. ### Task types Confluent Cloud for Apache Flink supports these types of analysis for AI model inference: - **Classification:** Categorize input data into predefined classes or labels. This task is used in applications like spam detection, where emails are classified as “spam” or “not spam”, and image recognition. - **Clustering:** Group a set of objects so that objects in the same group, called a “cluster”, are more similar to each other than to those in other groups. This task is a form of unsupervised learning, because it doesn’t rely on predefined categories. Applications include customer segmentation in marketing and gene sequence analysis in biology. - **Embedding:** Transform high-dimensional data into lower-dimensional vectors while preserving the relative distances between data points. This is crucial for tasks like natural language processing (NLP), where words or sentences are converted into vectors, enabling models to understand semantic similarities. Embeddings are used in recommendation systems, search engines, and more. - **Regression:** Regression models predict a continuous output variable based on one or more input features. This task is used in scenarios like predicting house prices based on features like size, location, and number of bedrooms, or forecasting stock prices. Regression analysis helps in understanding the relationships between variables and forecasting. - **Text generation:** Generate human-like text based on input data. Applications include chatbots, content creation, and language translation. When you register an AI or ML model, you specify the task type by using the [task](#flink-sql-create-model-with-task) property. `task` is a required property, but it applies only when using the [ML_EVALUATE](../functions/model-inference-functions.md#flink-sql-ml-evaluate-function) function. ## Examples The following code example shows how to run an AI model. The model must be created with the model provider and registered by using the CREATE MODEL statement with ``. ```sql SELECT * FROM my_table, LATERAL TABLE(ML_PREDICT('', column1, column2)); ``` All of the CREATE MODEL statements require a connection resource that you create by using the [CREATE CONNECTION](create-connection.md#flink-sql-create-connection) statement. For example, the following code example shows how to create a connection for AWS Bedrock. ```sql # Example command to create a connection for AWS Bedrock. CREATE CONNECTION bedrock_connection WITH ( 'type' = 'bedrock', 'endpoint' = 'https://bedrock-runtime.us-west-2.amazonaws.com/model/amazon.titan-embed-text-v1/invoke', 'aws-access-key' = '', 'aws-secret-key' = '', 'aws-session-token' = '' ); ``` ### Classification task The following example shows how to create an OpenAI classification model. For more information, see [Sentiment analysis example with an OpenAI model](../../../ai/ai-model-inference.md#flink-sql-ai-model-sentiment-analysis). ```sql CREATE MODEL sentimentmodel INPUT(text STRING) OUTPUT(sentiment STRING) COMMENT 'sentiment analysis model' WITH ( 'provider' = 'openai', 'task' = 'classification', '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.' ); ``` ### Clustering task The following example shows how to create an Azure ML clustering model. It requires that a K-Means model has been trained and deployed on Azure. Replace `` and `` with your values. ```sql CREATE MODEL clusteringmodel INPUT (vectors ARRAY, other_feature INT, other_feature2 STRING) OUTPUT (cluster_num INT) WITH ( 'task' = 'clustering', 'provider' = 'azureml', 'azureml.connection' = '' ); ``` ### Embedding task The following example shows how to create an AWS Bedrock text embedding model. Replace `` with your value. For more information, see [Text embedding examples](../../../ai/ai-model-inference.md#flink-sql-ai-model-text-embedding). ```sql CREATE MODEL embeddingmodel INPUT (text STRING) OUTPUT (embedding ARRAY) WITH ( 'task' = 'embedding', 'provider' = 'bedrock', 'bedrock.connection' = 'bedrock_connection' ); ``` ### Text generation task The following example shows how to create an OpenAI text generation task for translating from English to Spanish. ```sql CREATE MODEL translatemodel INPUT(english STRING) OUTPUT(spanish STRING) COMMENT 'spanish translation model' WITH ( 'provider' = 'openai', 'task' = 'text_generation', 'openai.connection' = '', 'openai.model_version' = 'gpt-3.5-turbo', 'openai.system_prompt' = 'Translate to spanish' ); ``` For more examples, see [Run an AI Model](../../../ai/ai-model-inference.md#flink-sql-ai-model). ## Model versioning A model can have multiple versions. A version is an integer number that starts at 1. The default version for a new model is 1. Currently, the maximum number of supported versions is 10. New versions are created by the CREATE MODEL statement for the same model name. A new version increments the current maximum version by 1. To view the versions of a model, use the [DESCRIBE MODEL](describe.md#flink-sql-describe-model) statement. Only model options are versioned, which means input/output format and comments don’t change across versions. The statement fails if input format, output format, or comments change. For model options, model task changes are not permitted. The following code example shows the result of running CREATE MODEL twice with the same model name. ```sql CREATE MODEL `my-model` ... -- Output `my-model` with version 1 created. Default version: 1 CREATE MODEL `my-model` ... -- Output `my-model` with version 2 created. Default version: 1 ``` By default, version 1 is the default version when a model is first created. As more versions are created by the CREATE MODEL statement, you can change the default version by using the ALTER MODEL statement. The following example shows how to change the default version of an existing model. ```sql ALTER MODEL SET ('default_version'=''); ``` You can access a specific version of a model in queries by using the `$` syntax. If no version is specified, the default version is used. The following code examples show how to use a specific version of a model in a query. ```sql -- Use version 2 of the model. SELECT * FROM `my-table` LATERAL TABLE (ML_PREDICT('my-model$2', col1, col2)); -- Use the default version of the model. SELECT * FROM `my-table` LATERAL TABLE (ML_PREDICT('my-model', col1, col2)); ``` Use the `$` syntax to delete a specific version of a model: ```sql -- Delete a specific version of the model. DROP MODEL `$`; -- Delete all versions and the model. DROP MODEL `$all`; ``` The maximum version number is the next default version. If all versions are dropped, the whole model is deleted. To change the version of an existing model, use the [ALTER MODEL](alter-model.md#flink-sql-alter-model) statement. If no version is specified, the default version is changed. ```sql ALTER MODEL `$` SET ('k1'='v1', 'k2'='v2'); ``` ## WITH options Specify the details of your AI inference model by using the WITH clause. The following tables show the supported properties in the WITH clause. | Model Provider | Property | |-------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | [Common](#flink-sql-create-model-with-common) | - [{PROVIDER}.client_timeout](#flink-sql-create-model-with-client-timeout)
- [{PROVIDER}.connection](#flink-sql-create-model-with-connection)
- [{PROVIDER}.input_format](#flink-sql-create-model-with-input-format)
- [{PROVIDER}.input_content_type](#flink-sql-create-model-with-input-content-type)
- [{PROVIDER}.output_format](#flink-sql-create-model-with-output-format)
- [{PROVIDER}.output_content_type](#flink-sql-create-model-with-output-content-type)
- [{PROVIDER}.PARAMS.\*](#flink-sql-create-model-with-params)
- [{PROVIDER}.system_prompt](#flink-sql-create-model-with-system-prompt) | | [OpenAI](#flink-sql-create-model-with-openai-options) | - [openai.input_format](#flink-sql-create-model-with-openai-input-format)
- [openai.model_version](#flink-sql-create-model-with-openai-model-version) | | [Azure OpenAI](#flink-sql-create-model-with-azure-openai-options) | - [azureopenai.input_format](#flink-sql-create-model-with-azure-openai-input-format)
- [azureopenai.model_version](#flink-sql-create-model-with-azure-openai-model-version) | | [Azure ML](#flink-sql-create-model-with-azureml) | - [azureml.input_format](#flink-sql-create-model-with-azureml-input-format)
- [azureml.deployment_name](#flink-sql-create-model-with-azureml-deployment-name) | | [Bedrock](#flink-sql-create-model-with-bedrock) | - [bedrock.model_version](#flink-sql-create-model-with-bedrock-model-version) | | [Google AI](#flink-sql-create-model-with-googleai) | - [googleai.input_format](#flink-sql-create-model-with-googleai-input-format) | | [Sagemaker](#flink-sql-create-model-with-sagemaker) | - [sagemaker.custom_attributes](#flink-sql-create-model-with-sagemaker-custom-attribute)
- [sagemaker.enable_explanations](#flink-sql-create-model-with-sagemaker-enable-explanations)
- [sagemaker.inference_component_name](#flink-sql-create-model-with-sagemaker-inference-component-name)
- [sagemaker.inference_id](#flink-sql-create-model-with-sagemaker-inference-id)
- [sagemaker.input_content_type](#flink-sql-create-model-with-sagemaker-input-content-type)
- [sagemaker.output_content_type](#flink-sql-create-model-with-sagemaker-output-content-type)
- [sagemaker.target_container_hostname](#flink-sql-create-model-with-sagemaker-target-container-hostname)
- [sagemaker.target_model](#flink-sql-create-model-with-sagemaker-target-model)
- [sagemaker.target_variant](#flink-sql-create-model-with-sagemaker-target-variant) | | [Vertex AI](#flink-sql-create-model-with-vertexai) | - [vertexai.service_key](#flink-sql-create-model-with-vertexai-service-key)
- [vertexai.input_format](#flink-sql-create-model-with-vertexai-input-format) | | [Confluent](#flink-sql-create-model-with-confluent) | - [confluent.model](#flink-sql-create-model-with-confluent-model) | ### Connection resource Secrets must be set by using a connection resource that you create by using the [CREATE CONNECTION](create-connection.md#flink-sql-create-connection) statement. The connection resource securely contains the provider endpoint and secrets like the API key. For example, the following code example shows how to create a connection to OpenAI, named `openai_connection`. ```sql CREATE CONNECTION openai_connection WITH ( 'type' = 'openai', 'endpoint' = 'https://api.openai.com/v1/chat/completions', 'api-key' = '' ); ``` Specify the connection by name in the [{PROVIDER}.connection](#flink-sql-create-model-with-connection) property of the WITH clause. The environment, cloud, and region options in the [CREATE CONNECTION](create-connection.md#flink-sql-create-connection) statement must be the same as the compute pool which uses the connection. The following code example shows how to refer to the connection named `openai_connection` in the WITH clause: ```properties 'openai.connection' = 'openai_connection' ``` The maximum secret length is 4000 bytes, which is checked after the string is converted to bytes. ### Common properties The following properties are common to all of the model providers. #### {PROVIDER}.client_timeout Set the request timeout to the client endpoint. #### {PROVIDER}.connection Set the credentials for connecting to a model provider. Create the connection resource by using the [CREATE CONNECTION](create-connection.md#flink-sql-create-connection) statement. This property is required. #### {PROVIDER}.input_format Set the json, text, or binary input format used by the model. Each provider has a default value. This property is optional. For supported input formats, see [Text generation and LLM model formats](#flink-sql-create-model-text-llm-formats) and [Other formats](#flink-sql-create-model-other-formats). #### {PROVIDER}.input_content_type The HTTP content media type header to set when calling the model. The value is a [Media/MIME type](https://www.iana.org/assignments/media-types/media-types.xhtml). The default is chosen based on `input_format`. Usually, this property is required only for Sagemaker and Bedrock models. #### {PROVIDER}.output_format Set the json, text, or binary output format used by the model. The default is chosen based on `input_format`. This property is optional. For supported output formats, see [Text generation and LLM model formats](#flink-sql-create-model-text-llm-formats) and [Other formats](#flink-sql-create-model-other-formats). #### {PROVIDER}.output_content_type The HTTP `Accept` media type header to set when calling the model. The value is a [Media/MIME type](https://www.iana.org/assignments/media-types/media-types.xhtml). The default is chosen based on `output_format`. Usually, this property is required only for Sagemaker and Bedrock models. #### {PROVIDER}.PARAMS.\* Provide parameters based on the `input_format`. The maximum number of parameters you can set is 32. This property is optional. For more information, see [Parameters](#flink-sql-create-model-format-params). #### {PROVIDER}.system_prompt A system prompt passed to an LLM model to give it general behavioral instructions. The value is a string. Not all models support a system prompt. This property is optional. #### task Specify the kind of analysis to perform. Supported values are: - “classification” - “clustering” - “embedding” - “regression” - “text_generation” This property is required, but it applies only when using the [ML_EVALUATE](../functions/model-inference-functions.md#flink-sql-ml-evaluate-function) function. ### OpenAI properties #### openai.input_format Set the input format used by the model. The default is `OPENAI-CHAT`. This property is optional. #### openai.model_version Set the version string of the requested model. The default is `gpt-3.5-turbo`. This property is optional. ### Azure OpenAI properties Properties for OpenAI models deployed in Azure AI Studio. Azure OpenAI accepts all of the [OpenAI parameters](#flink-sql-create-model-with-openai-options), but with a different endpoint. #### azureopenai.input_format Set the input format used by the model. The default is `OPENAI-CHAT`. This property is optional. #### azureopenai.model_version Set the version string of the requested model. The default is `gpt-3.5-turbo`. This property is optional. ### Azure ML properties Properties for both Azure Machine Learning and LLM models from Azure AI Studio can use this provider. #### azureml.input_format Set the input format used by the model. The default is `AZUREML-PANDAS-DATAFRAME`. For AI Studio LLMs, `OPENAI-CHAT` is usually the correct format, even for non-OpenAI models. This property is optional. #### azureml.deployment_name Set the model name. ### Bedrock properties The default value for the `input_format` property is determined automatically based on the model endpoint, or `AMAZON-TITAN-TEXT` if there is no match. If necessary, set the property to match the model for your endpoint. Bedrock determines which model to invoke from the model ID or Amazon Resource Name (ARN) in the connection’s `endpoint` property, not from any property in the CREATE MODEL statement. To use a different model or model version, update the existing connection’s `endpoint` value, or create a new connection with the corresponding `endpoint` value and set the model’s `bedrock.connection` property to reference it. #### bedrock.model_version For Anthropic Claude models invoked through Bedrock, set the version of the Anthropic API request schema to use. This property defaults to `bedrock-2023-05-31`, and you usually don’t need to set it yourself. This property doesn’t select which model or model version Bedrock invokes. Setting it to a model ID or ARN has no effect. Use the connection’s `endpoint` property instead. This property is optional and applies only to Anthropic models. ### Google AI properties #### googleai.input_format Set the input format used by the model. The default is `GEMINI-GENERATE`. This property is optional. ### Sagemaker properties #### sagemaker.custom_attributes Set a model-dependent value that is passed through to Sagemaker in the header of the same name. This property is optional. #### sagemaker.enable_explanations Enable writing explanations, if your model supports them. Passed through to Sagemaker in the header of the same name. If your model supports writing explanations, they should be disabled, because Confluent Cloud for Apache Flink currently doesn’t support reading them. Don’t set `enable_explanations` if the model doesn’t support explanations, because this causes Sagemaker to return an error. This property is optional. #### sagemaker.inference_component_name Specify which inference component to use in the endpoint. Passed through to Sagemaker in the header of the same name. This property is optional. #### sagemaker.inference_id Set an ID that is passed through to Sagemaker in the header of the same name. Used for tracking request origins. This property is optional. #### sagemaker.input_content_type The HTTP content media type header to set when calling the model. Setting this property overrides the `Content-type` header for the model request. Many Sagemaker models use this header to determine their behavior, but set it only if choosing an appropriate `input_format` is not sufficient. This property is optional. #### sagemaker.output_content_type The HTTP `Accept` media type header to set when calling the model. Setting this property overrides the `Accept` header for the model request. Some Sagemaker models use this header to determine their outputs, but set it only if choosing an appropriate `output_format` is not sufficient. This property is optional. #### sagemaker.target_container_hostname Allows calling a specific container when the endpoint has multiple containers. Passed through to Sagemaker in the header of the same name. This property is optional. #### sagemaker.target_model Enables calling a specific model from multiple models deployed to the same endpoint. Passed through to Sagemaker in the header of the same name. This property is optional. #### sagemaker.target_variant Enables calling a specific version of the model from multiple deployed variants. Passed through to Sagemaker in the header of the same name. This property is optional. ### Vertex AI properties #### vertexai.service_key Set the Service Account Key of a service account with permission to call the inference endpoint. This value is a secret. This property is required. #### vertexai.input_format Set the input format used by the model. The default is `TF-SERVING`. Defaults to `GEMINI-GENERATE` if the endpoint is for a published Gemini model. This property is optional. ### Confluent managed model properties #### NOTE > Managed AI models are an Early Access Program feature in Confluent Cloud. > > > > > > > > > > An Early Access feature is a component of Confluent Cloud introduced to gain > feedback. This feature should be used only for evaluation and non-production > testing purposes or to provide feedback to Confluent, particularly as it > becomes more widely available in follow-on preview editions. > Early Access Program features are intended for evaluation use in development > and testing environments only, and not for production use. Early Access Program > features are provided: (a) without support; (b) “AS IS”; and (c) without > indemnification, warranty, or condition of any kind. No service level commitment > will apply to Early Access Program features. Early Access Program features are > considered to be a Proof of Concept as defined in the Confluent Cloud Terms of Service. > Confluent may discontinue providing preview releases of the Early Access > Program features at any time in Confluent’s sole discretion. If you would like to participate in the Early Access Program, [sign up here](https://events.confluent.io/early-access-flink-features). #### confluent.model Set the managed model. The following values are supported. - [BAAI/bge-large-en-v1.5 (embedding)](#flink-sql-create-model-managed-baai-bge-large-en-v1-5) - [google/gemma-2-2b-it (LLM)](#flink-sql-create-model-managed-google-gemma-2-2b-it) - [intfloat/e5-base-v2 (embedding)](#flink-sql-create-model-managed-intfloat-e5-base-v2) - [meta-llama/Llama-3.1-8B-Instruct (LLM)](#flink-sql-create-model-managed-meta-llama-llama-3-1-8b-instruct) - [microsoft/Phi-3.5-mini-instruct (LLM)](#flink-sql-create-model-managed-microsoft-phi-3-5-mini-instruct) - [Qwen/Qwen2.5-7B-Instruct (LLM)](#flink-sql-create-model-managed-qwen-qwen2-5-7b-instruct) #### BAAI/bge-large-en-v1.5 (embedding) The following code example shows how to create an embedding with the `BAAI/bge-large-en-v1.5` managed model. ```sql CREATE MODEL `managed_model_embedding` INPUT (text STRING) OUTPUT (embedding ARRAY) WITH ( 'provider' = 'confluent', 'task' = 'embedding', 'confluent.model'='BAAI/bge-large-en-v1.5' ); ``` #### google/gemma-2-2b-it (LLM) The following code example shows how to create an LLM with the `google/gemma-2-2b-it` managed model. ```sql CREATE MODEL `managed_model_llm` INPUT (text STRING) OUTPUT (response STRING) WITH ( 'provider' = 'confluent', 'task' = 'classification', 'confluent.model'='google/gemma-2-2b-it' ); ``` #### intfloat/e5-base-v2 (embedding) The following code example shows how to create an embedding with the `intfloat/e5-base-v2` managed model. ```sql CREATE MODEL `managed_model_embedding` INPUT (text STRING) OUTPUT (embedding ARRAY) WITH ( 'provider' = 'confluent', 'task' = 'embedding', 'confluent.model'='intfloat/e5-base-v2' ); ``` #### meta-llama/Llama-3.1-8B-Instruct (LLM) The following code example shows how to create an LLM with the `meta-llama/Llama-3.1-8B-Instruct` managed model. ```sql CREATE MODEL `managed_model_llm` INPUT (text STRING) OUTPUT (response STRING) WITH ( 'provider' = 'confluent', 'task' = 'classification', 'confluent.model'='meta-llama/Llama-3.1-8B-Instruct' ); ``` #### microsoft/Phi-3.5-mini-instruct (LLM) The following code example shows how to create an LLM with the `microsoft/Phi-3.5-mini-instruct` managed model. ```sql CREATE MODEL `managed_model_llm` INPUT (text STRING) OUTPUT (response STRING) WITH ( 'provider' = 'confluent', 'task' = 'classification', 'confluent.model'='microsoft/Phi-3.5-mini-instruct' ); ``` #### Qwen/Qwen2.5-7B-Instruct (LLM) The following code example shows how to create an LLM with the `Qwen/Qwen2.5-7B-Instruct` managed model. ```sql CREATE MODEL `managed_model_llm` INPUT (text STRING) OUTPUT (response STRING) WITH ( 'provider' = 'confluent', 'task' = 'classification', 'confluent.model'='Qwen/Qwen2.5-7B-Instruct' ); ``` ## Supported input/output formats The following input/output formats for text generation and LLM models are supported. | [AI-21-COMPLETE](#flink-sql-create-model-ai-21-complete-format) | [AMAZON-TITAN-EMBED](#flink-sql-create-model-anthropic-amazon-titan-embed-format) | [AMAZON-TITAN-TEXT](#flink-sql-create-model-anthropic-amazon-titan-format) | |-------------------------------------------------------------------------------|-------------------------------------------------------------------------------------|------------------------------------------------------------------------------| | [ANTHROPIC-COMPLETIONS](#flink-sql-create-model-anthropic-completions-format) | [ANTHROPIC-MESSAGES](#flink-sql-create-model-anthropic-messages-format) | [AZURE-EMBED](#flink-sql-create-model-azure-embed-format) | | [BEDROCK-LLAMA](#flink-sql-create-model-bedrock-llama-format) | [COHERE-CHAT](#flink-sql-create-model-cohere-chat-format) | [COHERE-EMBED](#flink-sql-create-model-cohere-embed-format) | | [COHERE-GENERATE](#flink-sql-create-model-cohere-generate-format) | [GEMINI-GENERATE](#flink-sql-create-model-gemini-generate-format) | [GEMINI-CHAT](#flink-sql-create-model-gemini-chat-format) | | [MISTRAL-CHAT](#flink-sql-create-model-mistral-chat-format) | [MISTRAL-COMPLETIONS](#flink-sql-create-model-mistral-completions-format) | [OPENAI-CHAT](#flink-sql-create-model-openai-chat-format) | | [OPENAI-EMBED](#flink-sql-create-model-openai-embed-format) | [VERTEX-EMBED](#flink-sql-create-model-vertex-embed-format) | | The following additional input/output formats are supported. | [AZUREML-PANDAS-DATAFRAME](#flink-sql-create-model-azureml-pandas-dataframe-format) | [AZUREML-TENSOR](#flink-sql-create-model-azureml-tensor-format) | [BINARY](#flink-sql-create-model-binary-format) | |---------------------------------------------------------------------------------------|-----------------------------------------------------------------------|---------------------------------------------------------| | [CSV](#flink-sql-create-model-csv-format) | [JSON](#flink-sql-create-model-json-format) | [JSON-ARRAY](#flink-sql-create-model-json-array-format) | | [JSON:wrapper](#flink-sql-create-model-json-wrapper-format) | [KSERVE-V1](#flink-sql-create-model-kserve-v1-format) | [KSERVE-V2](#flink-sql-create-model-kserve-k2-format) | | [MLFLOW-TENSOR](#flink-sql-create-model-mlflow-tensor-format) | [PANDAS-DATAFRAME](#flink-sql-create-model-pandas-dataframe-format) | [TEXT](#flink-sql-create-model-text-format) | | [TF-SERVING](#flink-sql-create-model-tf-serving-format) | [TF-SERVING-COLUMN](#flink-sql-create-model-tf-serving-column-format) | [TRITON](#flink-sql-create-model-triton-format) | | [VERTEXAI-PYTORCH](#flink-sql-create-model-vertex-ai-pytorch-format) | | | ### Parameters The text generation and LLM formats support some or all of the following parameters. #### {PROVIDER}.PARAMS.temperature Controls the randomness or “creativity” of the output. Typical values are between 0.0 and 1.0. This parameter is model-dependent. Its type is `Float`. #### {PROVIDER}.PARAMS.top_p The probability cutoff for token selection. Usually, either [temperature](#flink-sql-create-model-temperature-param) or [top_p](#flink-sql-create-model-top-p-param) are specified, but not both. This parameter is model-dependent. Its type is `Float`. #### {PROVIDER}.PARAMS.top_k The number of possible tokens to sample from at each step. This parameter is model-dependent. Its type is `Float`. #### {PROVIDER}.PARAMS.stop A CSV list of strings to pass as stop sequences to the model. #### {PROVIDER}.PARAMS.max_tokens The maximum number of tokens for the model to return. Its type is `Int`. ### Text generation and LLM model formats The following formats are intended for text generation models and LLMs. They require that the model has a single STRING input and a single STRING output. #### AI-21-COMPLETE This format is for models using the [AI21 Labs J2 Complete API](https://docs.ai21.com/reference/j2-complete-ref), including the AI21 Labs Foundation models on AWS Bedrock. This format does not support the [top_k](#flink-sql-create-model-top-k-param) parameter. #### AMAZON-TITAN-EMBED This format is for [Amazon Titan Text Embedding](https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-titan-embed-text.html) models. #### AMAZON-TITAN-TEXT The format is for [Amazon’s Titan Text models](https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-titan-text.html). This is the default format for the AWS Bedrock provider. This format does not support the [top_k](#flink-sql-create-model-top-k-param) parameter. #### ANTHROPIC-COMPLETIONS This format is for models using the [Anthropic Claude Text Completions API](https://docs.anthropic.com/claude/reference/complete_post), including some Anthropic models on AWS Bedrock. #### ANTHROPIC-MESSAGES This format is for models using the [Anthropic Claude Messages API](https://docs.anthropic.com/claude/reference/messages_post), including some Anthropic models on AWS Bedrock. Some Anthropic models accept both this and the Completions API format. #### AZURE-EMBED The embedding format used by other foundation models on Azure. This format is the same as [OPENAI-EMBED](#flink-sql-create-model-openai-embed-format). #### BEDROCK-LLAMA The format used by [Llama models on AWS Bedrock](https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-meta.html). This format does not support the [top_k](#flink-sql-create-model-top-k-param) or [stop](#flink-sql-create-model-stop-param) parameters. #### COHERE-CHAT The [Cohere Chat API](https://docs.cohere.com/reference/chat) format. #### COHERE-EMBED Cohere’s [Embedding API](https://docs.cohere.com/reference/embed) format. #### COHERE-GENERATE The legacy [Cohere Chat API](https://docs.cohere.com/reference/generate) format. This format is used by AWS Bedrock Cohere Command models. #### GEMINI-GENERATE The [Google Gemini API](https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/gemini#gemini-1.0-pro) format. This is the default format for the Google AI provider, but you can also use it with Gemini models on the Google Vertex AI. #### GEMINI-CHAT Same as the [GEMINI-GENERATE](#flink-sql-create-model-gemini-generate-format) format. #### MISTRAL-CHAT The standard [Mistral API](https://docs.mistral.ai/api/) format. #### MISTRAL-COMPLETIONS The legacy [Mistral Completions API](https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-mistral.html) format used by AWS Bedrock. #### OPENAI-CHAT The [OpenAI Chat API](https://platform.openai.com/docs/api-reference/chat) format. This is the default for the OpenAI and Azure OpenAI providers. It is also generally used by most non-OpenAI LLM models deployed in Azure AI Studio using the Azure ML provider. #### OPENAI-EMBED The [OpenAI Embedding model](https://platform.openai.com/docs/guides/embeddings) format. #### VERTEX-EMBED The [Embedding format](https://cloud.google.com/vertex-ai/generative-ai/docs/model-reference/text-embeddings) for Vertex AI Gemini models. ### Other formats The following formats are intended for predictive models running on providers like Sagemaker, Vertex AI, and Azure ML. Usually, these models are used for tasks like classification, regression, and clustering. Currently, none of these formats support [PARAMS](#flink-sql-create-model-temperature-param). Unless specified, each input format defaults to the associated output format with the same name. #### AZUREML-PANDAS-DATAFRAME [Azure ML’s version](https://learn.microsoft.com/en-us/azure/machine-learning/how-to-deploy-mlflow-models#payload-example-for-a-json-serialized-pandas-dataframe-in-the-split-orientation) of the Pandas Dataframe Split format. The only difference is that this version has “input_data” as the top-level field, instead of “dataframe_split”. This is the default format for Azure ML models. The output format defaults to [JSON-ARRAY](#flink-sql-create-model-json-array-format). #### AZUREML-TENSOR [Azure ML’s version](https://learn.microsoft.com/en-us/azure/machine-learning/how-to-deploy-mlflow-models#payload-example-for-a-named-tensor-input) of named input tensors. Equivalent to the “JSON:input_data” input format. The output format defaults to “JSON:outputs”. #### BINARY Raw binary inputs, serialized in little-endian byte order. This input format accepts multiple input columns, which are packed in order. #### CSV Comma separated text. This is the default format for Sagemaker models, but Sagemaker models vary widely, and most models must choose a different format. #### JSON The inputs are formatted as a JSON object, with field names equal to the column names of the model input schema. The JSON format supports user-defined parameters. If you specify `'{provider}.params.some_key'='value'` in the WITH options, the key and value are used in the JSON input as `{"some_key": "value"}`. Example: ```json { "column1": "String Data", "column2": [1,2,3,4] } ``` #### JSON-ARRAY The inputs are formatted as a JSON array, including [] brackets, but without the {} braces of a top-level JSON object. Column names are not included in the format. If the model takes a single input array column, it will be output as the top-level array. Models with multiple inputs have their arrays nested in JSON fashion. This format is usually appropriate for models that expect Numpy arrays. Example: ```json [1,2,3,"String Data"] ``` #### JSON:wrapper Similar to the default JSON behavior, but all fields are wrapped in a named top-level object. The wrapper may be any valid JSON string. Example: ```json { "wrapper": { "column1": "String Data", "column2": [1,2,3,4] } } ``` #### KSERVE-V1 Same as the [TF-SERVING](#flink-sql-create-model-tf-serving-format) format. #### KSERVE-V2 Same as the [TRITON](#flink-sql-create-model-triton-format) format. #### MLFLOW-TENSOR The format used by some MLFlow models. It is the same format as [TF-SERVING-COLUMN](#flink-sql-create-model-tf-serving-column-format). #### PANDAS-DATAFRAME The [Pandas Dataframe Split](https://mlflow.org/docs/latest/deployment/deploy-model-locally.html#json-input) format used by most MLFlow models. The output format defaults to [JSON-ARRAY](#flink-sql-create-model-json-array-format). #### TEXT Model input values formatted as raw text. Use newlines to separate multiple inputs. #### TF-SERVING The [Tensorflow Serving Row](https://www.tensorflow.org/tfx/serving/api_rest#request_format_2) format. This is the default format for Vertex AI models. It is generally the correct format to use for most predictive models trained in Vertex AI. #### TF-SERVING-COLUMN The [TensorFlow Serving Column](https://www.tensorflow.org/tfx/serving/api_rest#specifying_input_tensors_in_column_format) format. It is exactly equivalent to “JSON:inputs”. The output format defaults to “JSON:outputs”. #### TRITON The [Triton/KServeV2](https://github.com/kserve/kserve/blob/master/docs/predict-api/v2/required_api.md) format used by NVidia Triton Inference Servers. When possible, this format serializes data in the protocol’s mixed json+binary format. Some Tensor datatypes, such as 16-bit floats, do not have an exact equivalent in Flink SQL, but they are converted when possible. #### VERTEXAI-PYTORCH [Vertex AI’s format for PyTorch models](https://cloud.google.com/vertex-ai/docs/predictions/get-online-predictions#request-body-details). This format is the [TF-SERVING](#flink-sql-create-model-tf-serving-format) format with an extra wrapper around the data. The output format defaults to [TF-SERVING](#flink-sql-create-model-tf-serving-format). ## Related content - [Run a managed AI model](../../../ai/managed-model.md#flink-sql-managed-ai-model) - [Run a remote AI Model](../../../ai/ai-model-inference.md#flink-sql-ai-model) - [ALTER MODEL](alter-model.md#flink-sql-alter-model) - [DROP MODEL](drop-model.md#flink-sql-drop-model) #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).