Realtime Compute for Apache Flink：通用調用

快速開始

前提條件

• 已建立 Flink 工作空間，詳情請參見開通Realtime ComputeFlink版。

• 已開通 Flink AI 服務，詳情請參見Flink AI服務（內建模型）。

• 使用 Flink AI 服務（內建模型）時，需使用 VVR 11.7 及以上版本引擎。

以下樣本示範如何使用 ML_PREDICT 調用 Flink 內建模型。請在資料開發 > ETL頁面，新增作業、複製代碼後點擊調試。。

CREATE TEMPORARY TABLE text_source (
  user_input STRING
) WITH ('connector' = 'datagen');

CREATE TEMPORARY TABLE result_sink (
  user_input STRING,
  ai_analysis STRING
) WITH ('connector' = 'print');

CREATE TEMPORARY MODEL text_model
INPUT (user_input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.6-flash',
  'task' = 'chat/completions',
  'system-prompt' = '分析內容亂碼程度，給0-100的評分'
);

INSERT INTO result_sink
SELECT user_input, content as ai_analysis FROM
    ML_PREDICT(
        TABLE text_source,
        MODEL text_model,
        DESCRIPTOR(user_input)
);

使用限制

• 僅Realtime Compute引擎 VVR 11.1 及以上版本支援。

• 部分參數僅在使用 Flink AI 服務（內建模型）時支援，僅支援 VVR 11.8.preview.2 及以上版本。

• ML_PREDICT 運算元的輸送量受百鍊平台限流限制。觸及資料傳輸量上限時，作業會出現以 ML_PREDICT 運算元為瓶頸的反壓現象，嚴重時可能觸發逾時報錯及作業重啟。詳情參見百鍊平台限流。

• content-types 中的類型數量及 Descriptor 中的列數，必須與 CREATE MODEL 中 INPUT 的列數一致。

• image_url 類型的列必須為 STRING，multi_image_urls 類型的列必須為 ARRAY<STRING>。

• base64 圖片必須包含 data:image/<格式>;base64, 首碼，不支援純 base64 字串和本地檔案路徑。

文法

ML_PREDICT(TABLE <table_name>, MODEL <model_name>, DESCRIPTOR(<input_columns>) [, CONFIG => MAP[...]])        

入參

調用級參數

使用者可以在使用 ML_PREDICT 時指定調用級參數。若 CREATE MODEL 中已設定參數值，則調用 ML_PREDICT 時以新指定的參數為準，但不會持久化改寫 MODEL 中的參數值。

樣本

-- 註冊引用內建模型，不指定 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', '請用英文回複']);

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

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

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

調用時指定參數

樣本 1：指定內容類型參數

-- CREATE MODEL 時未指定 content-type / content-types，預設 content-type = text
CREATE MODEL model_single
INPUT (input STRING)
OUTPUT (content STRING)
WITH (
  'provider' = 'openai-compat',
  'model' = 'qwen3.5-plus',
  'task' = 'chat/completions'
);

-- 調用 1：沿用 MODEL 參數，內容類型為 text
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_text, 
  MODEL model_single, 
  DESCRIPTOR(input)
));

-- 調用 2：調用時指定內容類型為圖片
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_img, 
  MODEL model_single, 
  DESCRIPTOR(input),
  MAP['content-type', 'image_url']
));

樣本 2：指定 user-prompt 及列級參數

-- CREATE MODEL 時設定預設配置（多列模型）
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',
  'user-prompt' = '請描述圖片內容'
);

-- 調用 1：使用預設配置
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_a, MODEL vl_model, DESCRIPTOR(text_input, image_input)
));

-- 調用 2：覆蓋 user-prompt 和列級參數
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_b, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP[
    'user-prompt', '請用英文回答',
    'image_input.min_pixels', '100',
    'image_input.max_pixels', '5000'
  ]
));

-- 調用 3：覆蓋 content-types，改變類型組合（兩列都當文本處理）
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_c, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP['content-types', 'text;text']
));

-- 調用 4：覆蓋 content-types，改變類型組合（翻轉順序）
SELECT content FROM TABLE(ML_PREDICT(
  TABLE source_c, MODEL vl_model, DESCRIPTOR(text_input, image_input),
  MAP['content-types', 'image_url;text']
));