General invocation

Content type parameters

• For single-column input, use content-type to specify the content type. Supported values: text, image_url.

• For multi-column input, use content-types to specify the content type for each column. The supported values and their mapping to Flink SQL column types are listed below:

• We recommend that you specify content-types in the parameters. If you set both content-type and content-types, the framework reports an error. Note: the content-type and content-types settings in model parameters and per-call settings affect each other:

Column-level parameters

Bundle processing

• Processing: Bundle processing first aggregates multiple input records into a batch and then calls the AI model for the batch. This reduces the number of model calls but increases the model inference time for each request. When records can be effectively aggregated, bundle processing can significantly improve operator throughput.

• Parameter description: bundle-size specifies the maximum number of records accumulated in a batch. bundle-allow-latency specifies the maximum wait time when a batch has not reached the bundle-size value. Processing starts when either the batch-size or wait-time condition is met.

• Data type: Only text is supported.

• Supported functions: ML_PREDICT, text classification (AI_CLASSIFY), sentiment analysis (AI_SENTIMENT), information extraction (AI_EXTRACT), text summarization (AI_SUMMARIZE), text translation (AI_TRANSLATE), data masking (AI_MASK), and text embedding (AI_EMBED).

• Global configuration: This configuration applies to every ML_PREDICT call and every call to the listed domain-specific AI functions in the job. Use the following global parameters to enable bundle processing:

  SET 'table.exec.ml-predict.bundle-size' = '10';
  SET 'table.exec.ml-predict.bundle-allow-latency' = '5s';

• Per-call configuration: These parameters apply only to the function call in which they are specified. ML_PREDICT and the listed domain-specific AI functions use the same configuration method. For parameters and examples, see Per-call configuration.

Text

The following example registers and uses a Flink AI Service built-in model to classify input text by sentiment. Example 1 uses MODEL-level parameters. Example 2 overrides user-prompt at call time.

-- Register a built-in model. When content-type is not specified, it defaults to text
CREATE MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'system-prompt' = 'You are a sentiment classifier. Output one label: negative, positive, or neutral'
);

-- Create a source table: simulated product review data
CREATE TEMPORARY VIEW input_table(id, content)
AS VALUES
  (1, 'Great quality, soft fabric, fits perfectly'),
  (2, 'Had loose threads on arrival, faded badly after one wash'),
  (3, 'Received the item, looks as pictured'),
  (4, 'Started pilling after two weeks, customer service refused returns'),
  (5, 'Flattering fit, color is even better than the photo, already ordered a third one');

-- Create a result table
CREATE TEMPORARY TABLE output_table (
  id INT,
  content STRING,
  sentiment STRING
) WITH (
  'connector' = 'print'
);

-- Use ML_PREDICT for real-time inference
-- Example 1: Use MODEL-level parameters
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

-- Example 2: Specify per-call parameters
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content),
  MAP['user-prompt', 'Reply in Chinese']);

Example 1 output:

Example 2 output:

Image

The following example registers a multimodal model and classifies input images.

-- Register a built-in model with content-type set to image_url
CREATE TEMPORARY MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'content-type' = 'image_url'
);

-- Use ML_PREDICT for real-time inference
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Video

The following example registers a multimodal model to understand an input video.

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url',
  'system-prompt' = 'Briefly describe the video content'
);

CREATE TEMPORARY VIEW input_table(content)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content));

Text + single image

The following example registers a multimodal model for chat-based inference on text and image inputs.

CREATE MODEL vl_model
INPUT (text_input STRING, image_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url'
);

INSERT INTO result_sink
SELECT content FROM TABLE(ML_PREDICT(
  TABLE image_source,
  MODEL vl_model,
  DESCRIPTOR(text_input, image_input)
));         

Text + multiple images (multiple columns)

Pass each image through a separate INPUT column. Specify image_url for each image column in content-types.

CREATE MODEL vl_model_multi
INPUT (prompt STRING, img1 STRING, img2 STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;image_url;image_url'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_multi,
  DESCRIPTOR(prompt, img1, img2)
));          

Text + multiple images (array)

Pass multiple images in an ARRAY<STRING> column. Use multi_image_urls in content-types.

CREATE MODEL vl_model_array
INPUT (prompt STRING, images ARRAY<STRING>)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions',
  'content-types' = 'text;multi_image_urls'
);

SELECT content FROM TABLE(ML_PREDICT(
  TABLE my_source,
  MODEL vl_model_array,
  DESCRIPTOR(prompt, images)
)); 

Text + video

CREATE TEMPORARY MODEL sentiment_model
INPUT (video_url STRING, question STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-plus',
  'content-types' = 'video_url;text'
);

CREATE TEMPORARY VIEW input_table(content, question)
AS VALUES ('https://help-static-aliyun-doc.aliyuncs.com/file-manage-files/zh-CN/20251111/ckgcxt/%E5%8D%95%E4%BA%BA%E8%AF%B4%E8%AF%9D-1.mp4', 'How many people are there?');

SELECT
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table,
  MODEL sentiment_model,
  DESCRIPTOR(content, question));