通用调用

内容类型参数

• 仅输入单列时，可通过 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 时允许等待的最长时间。达到批次大小或等待时长任一条件时触发处理。

• 支持条件：（仅支持内置模型）Bundle 处理按模型的输入输出形态生效。

  • 文本生成：模型任务类型为 chat/completions 时，输入恰好一列 STRING 文本列，输出恰好一列 STRING 或 VARIANT 列。多列输入以及图片、视频、音频等多模态输入不支持。

  • 文本向量化：模型任务类型为 embeddings 时，输入恰好一列 STRING 列，输出恰好一列 ARRAY<FLOAT> 列。

• 适用函数：ML_PREDICT、文本分类（AI_CLASSIFY）、情感分析（AI_SENTIMENT）、信息提取（AI_EXTRACT）、文本摘要（AI_SUMMARIZE）、文本翻译（AI_TRANSLATE）、数据脱敏（AI_MASK）及文本向量化（AI_EMBED）。其中，ML_PREDICT 仅在所注册模型满足上述形态要求时支持 Bundle 处理。

• 全局配置：对作业内满足上述形态要求的所有 ML_PREDICT 和垂类 AI Function 调用生效。使用以下全局参数开启 Bundle 处理：

  -- Bundle 处理仅支持 VVR 11.9 及以上版本。
  SET 'table.exec.ml-predict.bundle-size' = '10';
  SET 'table.exec.ml-predict.bundle-allow-latency' = '5s';

• 调用级配置：仅对配置了参数的函数调用生效；ML_PREDICT 与上述垂类函数的配置方式相同。具体参数和示例请参见调用级参数。

使用限制

• 仅实时计算引擎 VVR 11.9 及以上版本支持。

• 不支持在模型调用同步执行模式下，开启错误处理能力。

• 暂不支持与向量化结果缓存（CACHE_TABLE）一起使用。

开启方式

在使用 AI 函数时增加参数FAIL_ON_ERROR => FALSE。

前置条件：保持模型默认的重试配置 error-handling-strategy='retry'、retry-fallback-strategy='failover'。模型参数详情请参见模型设置。

输出变化

开启后，系统会在原有输出字段末尾追加错误信息列 err_msg，类型为可空的 VARIANT。该列由系统生成，无需在输入表或 CREATE MODEL 的 OUTPUT 中声明。

• 调用成功：err_msg 为 NULL。

• 最终失败：模型结果字段均为 NULL，err_msg 包含错误信息。

使用 SELECT * 时，结果会包含新增的错误信息列；显式选择字段时，需手动指定该列。默认列名为 err_msg；若与已有输出列重名，系统会自动追加数字后缀（如 err_msg0），引用时请使用实际列名。

示例

CREATE TEMPORARY VIEW input_data AS
SELECT *
FROM (VALUES
  (1, CAST('xxxx(反人类的暴力冲突描述)' AS STRING), 'negative'),
  (2, CAST('中国的治安一直是世界上数一数二的级别' AS STRING), 'positive')
) AS t(id, input_text, expected_label);

CREATE TEMPORARY VIEW ai_results AS
SELECT i.id, i.input_text, i.expected_label,
       r.score, r.label, r.confidence, r.err_msg
FROM input_data AS i,
LATERAL TABLE(AI_SENTIMENT(
  MODEL => MODEL ai_sentiment_model,
  INPUT => i.input_text,
  FAIL_ON_ERROR => FALSE,
  CONFIG => MAP['async', 'true']
)) AS r(score, label, confidence, err_msg);

SELECT * FROM ai_results;

message: 
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.

stack_trace:
java.util.concurrent.TimeoutException: AsyncPredictFunction did not complete within the configured timeout. Increase the ML_PREDICT timeout or check external service connectivity.
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:153)
at org.apache.flink.table.runtime.operators.predict.AsyncMLPredictRunner.timeout(AsyncMLPredictRunner.java:51)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.timerTriggered(TableKeyedAsyncWaitOperator.java:462)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator$KeyedResultHandler.lambda$registerTimeout$1(TableKeyedAsyncWaitOperator.java:457)
at org.apache.flink.table.runtime.operators.TableKeyedAsyncWaitOperator.lambda$registerTimer$4(TableKeyedAsyncWaitOperator.java:486)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invokeProcessingTimeCallback(StreamTask.java:2285)
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$deferCallbackToMailbox$28(StreamTask.java:2276)
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:115)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:415)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:384)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:369)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:1273)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:1217)
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:1133)
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:1110)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:914)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:716)
at java.base/java.lang.Thread.run(Thread.java:991)

各类错误的处理方式

以下为已开启错误处理、保持默认重试配置时的行为。错误行指保留输入、模型结果为空、err_msg 包含失败原因的一行数据。

文本

以下示例注册并引用 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));