实时计算Flink版：通用调用

快速开始

前提条件

• 已创建 Flink 工作空间，详情请参见开通实时计算Flink版。

• 已开通 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)
);

使用限制

• 仅实时计算引擎 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']
));