Realtime Compute for Apache Flink：通用調用

內容類型參數

• 僅輸入單列時，可通過 content-type 指定內容類型，支援的值：text、image_url。

• 輸入多列時，可通過 content-types 指定每列內容類型，支援的值和 Flink SQL 列類型的映射關係：

• 請在參數中優先指定content-types，若同時設定 content-type 和 content-types 架構會報錯提示。注意：content-type 和 content-types在模型參數中設定和調用時設定是相互影響的：

列級參數

Bundle 處理

• 處理方式：Bundle 處理會先將多條輸入資料彙總成一個批次，再統一調用 AI 模型。這種處理方式可以減少模型調用次數，但會增加單次請求的模型推理時間，在資料被有效彙總時能明顯提升運算元輸送量。

• 參數說明：bundle-size 用於指定單個批次最多積累的資料條數；bundle-allow-latency 用於指定批次未達到 bundle-size 時允許等待的最長時間。達到批次大小或等待時間長度任一條件時觸發處理。

• 資料類型：僅支援 text。

• 適用函數：ML_PREDICT、文本分類（AI_CLASSIFY）、情感分析（AI_SENTIMENT）、資訊提取（AI_EXTRACT）、文本摘要（AI_SUMMARIZE）、文本翻譯（AI_TRANSLATE）、資料脫敏（AI_MASK）及文本向量化（AI_EMBED）。

• 全域配置：對作業內所有 ML_PREDICT 和上述垂類 AI Function 調用生效。使用以下全域參數開啟 Bundle 處理：

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

• 調用級配置：僅對配置了參數的函數調用生效；ML_PREDICT 與上述垂類函數的配置方式相同。具體參數和樣本請參見調用級參數。

文本

以下樣本註冊並引用 Flink AI 服務提供的內建模型，對輸入文本進行情感分類。調用樣本 1 複用 MODEL 級參數，調用樣本 2 在調用時指定參數 user-prompt。

-- 註冊引用內建模型，不指定 content-type 時預設為 text
CREATE MODEL sentiment_model
INPUT (prompt STRING)
OUTPUT (response STRING)
WITH (
  'provider' = 'openai-compat',
  'task' = 'chat/completions',
  'model' = 'qwen3.6-flash',
  'system-prompt' = '你是一個情感分類模型，輸出標籤：負面、正面、中性'
);

-- 建立源表：類比商品評價資料
CREATE TEMPORARY VIEW input_table(id, content)
AS VALUES 
  (1, '品質很好，面料柔軟親膚，尺碼也很標準，好評'),
  (2, '收到就有線頭，洗了一次嚴重褪色，不值這個價'),
  (3, '衣服已收到，和圖片一致'),
  (4, '穿了兩周起球嚴重，客服還不給退貨，差評'),
  (5, '版型顯瘦，顏色比圖片還好看，已經回購第三件了');

-- 建立結果表
CREATE TEMPORARY TABLE output_table (
  id INT,
  content STRING,
  sentiment STRING
) WITH (
  'connector' = 'print'
);

-- 使用 ML_PREDICT 進行即時推理
-- 調用樣本 1：複用 MODEL 級參數
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table, 
  MODEL sentiment_model, 
  DESCRIPTOR(content));

-- 調用樣本 2：指定調用級參數
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table, 
  MODEL sentiment_model, 
  DESCRIPTOR(content),
  MAP['user-prompt', '請用英文回複']);

調用樣本 1 輸出結果：

調用樣本 2 輸出結果：

圖片

以下樣本註冊一個多模態模型，對輸入圖片進行分類。

圖片輸入列（content-type為image_url）為STRING類型，支援以下兩種內容形式：

如果圖片儲存在使用Paimon Catalog的Blob列中，可以利用Paimon提供的Blob預簽名URL功能產生公網URL作為輸入，詳情請參見流式資料湖倉Paimon。

-- 註冊引用內建模型，指定 content-type 為 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'
);

-- 使用 ML_PREDICT 進行即時推理
INSERT INTO output_table
SELECT
  id,
  content,
  response AS sentiment
FROM ML_PREDICT(
  TABLE input_table, 
  MODEL sentiment_model, 
  DESCRIPTOR(content));

視頻

以下樣本註冊一個多模態模型，對輸入視頻進行理解。

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' = '請簡單描述下視頻內容'
);

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));

文本 + 單張圖片

以下樣本註冊一個多模態模型，對輸入的文本和圖片進行對話推理。

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)
));         

文本 + 多張圖片（多列）

通過多個 INPUT 列分別傳入每張圖片，content-types 中為每列指定 image_url。

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)
));          

文本 + 多張圖片（數組）

通過 ARRAY<STRING> 類型列傳入多張圖片，content-types 中使用 multi_image_urls。

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)
)); 

文本 + 視頻

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', '有幾個人');

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