实时计算 Flink 支持在 SQL 作业中调用外部多模态 Embedding 模型服务，将文本、图片、视频或图文组合转换为向量，为语义检索、以图搜图和多模态检索准备数据。您可以通过 CREATE MODEL 配置模型服务，再使用 ML_PREDICT 对输入数据进行推理，并保留业务字段与生成的向量一同输出。
本文介绍方舟 ARK、LAS 和 Context Search 多模态 Embedding 服务的接入方法。这三类服务均使用 provider = 'ark'，由 ARK 适配器调用兼容 ARK 多模态 Embedding 协议的接口。使用前，请查阅所选服务的模型说明，确认模型支持所需的输入类型，以及输出向量的维度。

• 准备 Flink SQL 作业。​使用 Flink 2.2-volcano 及以上引擎版本。
• 准备模型服务。​获取所选服务的完整请求地址（endpoint）、API Key 和模型标识，确认具有模型调用权限，并确认模型支持所需的媒体格式和向量维度。
• 配置网络访问。​确保 Flink 运行节点能够访问模型 endpoint；使用图片或视频时，模型服务还需能够访问媒体 URL。私有 TOS 对象应使用有效的预签名 HTTPS URL。
• 准备凭证。​通过平台加密变量配置 API Key。需要读取私有 TOS 对象时，还需准备具有对象读取权限的 TOS AK/SK。

执行示例前，请按以下约定准备配置：

• <...> 表示需要手动替换的配置占位符，例如将 <MULTIMODAL_ENDPOINT> 和 <MODEL_ID> 分别替换为实际服务地址和模型标识。
• ${secret_values.model-api-key} 等表达式表示平台加密变量引用。请先配置对应的加密变量，并将变量值设为实际凭证；使用已有变量时，将引用中的变量名改为实际名称。
• endpoint、API Key 和模型标识须对应同一服务。

定义模型 #

CREATE MODEL 是模型定义语句（DDL），用于在 SQL 中注册模型名称、输入输出结构及模型服务配置。本文用它接入已有的 Embedding 模型服务，后续通过 ML_PREDICT 调用服务生成向量。
CREATE TEMPORARY MODEL 是 CREATE MODEL 带可选关键字 TEMPORARY 的形式，用于注册临时模型定义。本文示例使用该形式供当前作业调用。完整语法请参见 Apache Flink CREATE MODEL；本文介绍的三类服务，其配置参数及输入输出要求见下文。

CREATE TEMPORARY MODEL embedding_model
INPUT (payload STRING)
OUTPUT (embedding ARRAY<FLOAT>)
WITH (
    'provider' = 'ark',
    'endpoint' = '<MULTIMODAL_ENDPOINT>',
    'api-key' = '${secret_values.model-api-key}',
    'model' = '<MODEL_ID>',
    'model.encoding-format' = 'float'
);

上述定义中，各部分的作用如下：

• embedding_model：在 Flink SQL 中注册的模型名称，后续通过 MODEL embedding_model 引用。
• INPUT (payload STRING)：声明模型接收一个字符串字段 payload，内容可以是文本或多模态 JSON。在 ML_PREDICT调用时，通过 DESCRIPTOR 指定输入表中实际用于推理的列，例如 DESCRIPTOR(payload)。
• OUTPUT (embedding ARRAY<FLOAT>)：声明模型返回一个浮点数组字段 embedding，用于存放生成的向量。
• WITH (...)：配置实际调用的模型服务，包括适配器、服务地址、凭证和模型标识。其中 'model' = '<MODEL_ID>' 指定服务端的模型标识，与 SQL 中注册的名称 embedding_model 含义不同。

接入本文介绍的方舟 ARK、LAS 和 Context Search 多模态 Embedding 服务时，CREATE MODEL 的 INPUT 必须声明且只声明一个 STRING 类型的输入字段，OUTPUT 必须声明且只声明一个 ARRAY<FLOAT> 类型的输出字段。输入输出字段名可以自定义。
定义模型后，可以通过ML_PREDICT调用该模型，通过DESCRIPTOR指定输入表或视图中的一个 STRING 类型字段，与模型的输入声明INPUT匹配。输入表中的其他业务字段可以在 SELECT 中保留。

CREATE TEMPORARY VIEW demo_input (id, payload) AS VALUES
    ('text-1', '红色运动鞋');

SELECT id, embedding
FROM ML_PREDICT(
    TABLE demo_input,
    MODEL embedding_model,
    DESCRIPTOR(payload)
);

• TABLE demo_input：指定提供数据的输入表或视图。
• MODEL embedding_model：引用前面通过 CREATE MODEL 定义的模型。
• DESCRIPTOR(payload)：选择输入数据中的 payload 字段，将其传给模型。
• SELECT id, embedding：保留输入中的业务 ID，并取出模型返回的向量。ML_PREDICT 的结果包含原输入表字段和向量字段，因此可以同时查询 id 和 embedding。

完整的 print 输出和日志查看步骤，请参见三类服务完整接入示例。

选择服务地址 #

根据所选服务配置完整 endpoint，包括接口路径以及必要的代理前缀和路由查询参数。切换服务时，同时调整 API Key 和模型标识。

• 方舟 ARK：​北京地域的多模态接口地址为 https://ark.cn-beijing.volces.com/api/v3/embeddings/multimodal。模型和 API Key 可在方舟控制台获取。
• LAS：​北京地域的接口地址为 https://operator.las.cn-beijing.volces.com/api/v1/embeddings/multimodal。使用 LAS API Key 和实际可用的模型标识，模型能力可参考LAS 图文向量化服务文档。
• Context Search：​使用 Context Search 推理服务提供的完整多模态地址，例如 https://<INFERENCE_HOST>/v1/embeddings/multimodal，并保留服务要求的代理前缀及路由查询参数。

配置参数 #

模型输入列使用 STRING 类型，可传入普通文本、单个内容对象的 JSON 或多个内容部分组成的 JSON 数组。

以 { 或 [ 开头的内容按 JSON 解析，JSON 格式错误或不支持的 type 会报错。普通 URL 字符串按文本处理；图片和视频推理需使用对应的 JSON 结构。
业务表中已有图片 URL 时，可使用 SQL JSON 函数或 IMAGE_URL_PAYLOAD
构造输入。私有 TOS 图片需先生成模型服务可读取的预签名 URL。相关函数的语法、参数和限制请参见IMAGE_URL_PAYLOAD。
所选服务和模型支持视频输入时，可将表格中的 video_url JSON 作为完整接入示例的 payload 输入。fps 指定抽帧频率；省略时为 1.0。
每条非空且推理成功的输入产生一个稠密浮点向量。可使用 CARDINALITY(embedding) 检查向量维度。SQL NULL 输入会跳过推理，不产生该行预测结果。

接入方舟 ARK #

本例使用方舟北京地域的多模态接口和 doubao-embedding-vision-251215 模型，将输出维度设为 2048，分别生成文本、图片和图文组合向量。请配置 ark-api-key 加密变量，并将图片地址替换为模型服务可访问的真实地址。

-- 1. 定义方舟模型，配置方舟 API Key 和模型标识
CREATE TEMPORARY MODEL ark_embedding
INPUT (payload STRING)
OUTPUT (embedding ARRAY<FLOAT>)
WITH (
    'provider' = 'ark',
    'endpoint' = 'https://ark.cn-beijing.volces.com/api/v3/embeddings/multimodal',
    'api-key' = '${secret_values.ark-api-key}',
    'model' = 'doubao-embedding-vision-251215',
    -- 本例指定 2048 维；使用其他模型时需核对其支持的维度
    'model.dimensions' = '2048',
    'model.encoding-format' = 'float'
);

-- 2. 准备三条输入，分别生成文本、图片和图文组合向量
CREATE TEMPORARY VIEW ark_input (id, payload) AS VALUES
    ('text-1', '红色运动鞋'),
    -- 将图片地址替换为模型服务可访问的真实地址
    ('image-1', '{"type":"image_url","image_url":{"url":"https://<MEDIA_HOST>/demo.jpg"}}'),
    -- JSON 数组中的文本和图片组成一次推理，返回一个组合向量
    ('mixed-1', '[{"type":"text","text":"红色运动鞋"},{"type":"image_url","image_url":{"url":"https://<MEDIA_HOST>/demo.jpg"}}]');

-- 3. 创建临时 print 表，输出业务 ID、向量及其实际维度
CREATE TEMPORARY TABLE ark_embedding_print (
    id STRING,
    embedding ARRAY<FLOAT>,
    dimensions INT
) WITH (
    'connector' = 'print'
);

-- 4. 调用本例定义的方舟模型；CARDINALITY 返回向量数组的长度
INSERT INTO ark_embedding_print
SELECT id, embedding, CARDINALITY(embedding) AS dimensions
FROM ML_PREDICT(
    TABLE ark_input,
    MODEL ark_embedding,
    DESCRIPTOR(payload)
);

请求均成功时，可在 TaskManager 的 Stdout 日志中查看三行输出，分别对应 text-1、image-1 和 mixed-1；每行包含业务 ID、浮点向量和实际维度。若服务返回 does not support matryoshka representation，请移除 model.dimensions 后重试，并按实际输出维度调整下游配置。

接入 LAS Embedding #

本例使用 LAS 北京地域的多模态接口。请配置 las-api-key 加密变量，并将 <LAS_MODEL_ID> 替换为 LAS 中可调用且支持文本和图片输入的模型标识。其他地域请使用对应的实际服务地址。模型 ID、支持的维度和媒体格式以 LAS 服务为准。具体可参考 LAS AI 服务页面。

-- 1. 定义 LAS 模型；兼容 ARK 多模态协议的服务仍使用 ark Provider
CREATE TEMPORARY MODEL las_embedding
INPUT (payload STRING)
OUTPUT (embedding ARRAY<FLOAT>)
WITH (
    'provider' = 'ark',
    'endpoint' = 'https://operator.las.cn-beijing.volces.com/api/v1/embeddings/multimodal',
    'api-key' = '${secret_values.las-api-key}',
    'model' = '<LAS_MODEL_ID>',
    'model.encoding-format' = 'float'
);

-- 2. 准备文本和图片两条输入；所选 LAS 模型需支持这两类输入
CREATE TEMPORARY VIEW las_input (id, payload) AS VALUES
    ('text-1', '红色运动鞋'),
    -- 将图片地址替换为模型服务可访问的真实地址
    ('image-1', '{"type":"image_url","image_url":{"url":"https://<MEDIA_HOST>/demo.jpg"}}');

-- 3. 创建临时 print 表，输出业务 ID、向量及其实际维度
CREATE TEMPORARY TABLE las_embedding_print (
    id STRING,
    embedding ARRAY<FLOAT>,
    dimensions INT
) WITH (
    'connector' = 'print'
);

-- 4. 调用本例定义的 LAS 模型，并输出实际向量维度
INSERT INTO las_embedding_print
SELECT id, embedding, CARDINALITY(embedding) AS dimensions
FROM ML_PREDICT(
    TABLE las_input,
    MODEL las_embedding,
    DESCRIPTOR(payload)
);

请求均成功时，可在 TaskManager 的 Stdout 日志中查看两行输出，分别包含文本和图片的业务 ID、向量及实际维度。
使用私有 TOS 图片时，请先生成模型服务可读取的预签名 HTTPS URL，再填入 image_url.url；有效期应覆盖作业排队、重试和模型读取图片的时间。

接入 Context Search 推理服务 #

请配置 context-search-api-key 加密变量，将 <CONTEXT_SEARCH_MODEL_ID> 替换为该推理服务中的模型标识，并将 endpoint 的完整值替换为服务提供的多模态请求地址。地址中的代理前缀及路由查询参数应一并保留。本例未指定 model.dimensions，输出维度由所选模型决定。

-- 1. 定义 Context Search 模型，配置该推理服务的凭证及模型标识
CREATE TEMPORARY MODEL context_search_embedding
INPUT (payload STRING)
OUTPUT (embedding ARRAY<FLOAT>)
WITH (
    'provider' = 'ark',
    -- 将整个 endpoint 值替换为服务提供的完整地址，保留代理前缀和路由查询参数
    'endpoint' = 'https://<INFERENCE_HOST>/v1/embeddings/multimodal',
    'api-key' = '${secret_values.context-search-api-key}',
    'model' = '<CONTEXT_SEARCH_MODEL_ID>',
    'model.encoding-format' = 'float'
);

-- 2. 准备一条待向量化的文本输入
CREATE TEMPORARY VIEW context_search_input (id, payload) AS VALUES
    ('text-1', '介绍一下实时计算 Flink');

-- 3. 创建临时 print 表，输出业务 ID、向量及其实际维度
CREATE TEMPORARY TABLE context_search_embedding_print (
    id STRING,
    embedding ARRAY<FLOAT>,
    dimensions INT
) WITH (
    'connector' = 'print'
);

-- 4. 调用本例定义的 Context Search 模型，并输出实际向量维度
INSERT INTO context_search_embedding_print
SELECT id, embedding, CARDINALITY(embedding) AS dimensions
FROM ML_PREDICT(
    TABLE context_search_input,
    MODEL context_search_embedding,
    DESCRIPTOR(payload)
);

请求成功时，可在 TaskManager 的 Stdout 日志中查看一行输出：id 为 text-1，embedding 为模型生成的浮点数组，dimensions 为实际向量维度。

以下示例通过 TOS-CDC 读取图片对象信息，构造私有图片的模型输入，再调用方舟 ARK 生成向量，最后将对象信息和向量输出到临时 print 表。

准备资源 #

• 准备存放图片的 TOS 目录，确认目录内对象的格式受所选模型支持，并配置 TOS-CDC 所需的变更通知和 Kafka 接入信息。
• 准备方舟多模态 Embedding 模型，配置 ark-api-key 加密变量。
• 配置 tos-ak 和 tos-sk 加密变量，确保凭证具有目标对象读取权限；TOS 地域和 Endpoint 与对象所在地域一致。

编写 SQL #

替换示例中的 TOS 路径和 Kafka 接入信息。示例使用北京地域；其他地域请同步调整 TOS 地域和 Endpoint。
本例使用 doubao-embedding-vision-251215，并将输出向量维度设为 2048。使用其他模型时，请替换模型标识，并根据该模型支持的维度调整或移除 model.dimensions。

-- 1. 定义 TOS-CDC 来源表，读取对象信息；请替换 TOS 路径和 Kafka 配置
CREATE TEMPORARY TABLE object_changes (
    object_key STRING NOT NULL,
    object_url STRING,
    bucket_name STRING,
    file_name STRING,
    object_etag STRING,
    object_size BIGINT,
    mtime TIMESTAMP_LTZ(3),
    event_time TIMESTAMP_LTZ(3),
    record_origin STRING
) WITH (
    'connector' = 'tos-cdc',
    'path' = 'tos://<BUCKET>/<IMAGE_PREFIX>/',
    'topic' = '<TOS_CDC_TOPIC>',
    'properties.bootstrap.servers' = '<KAFKA_BOOTSTRAP_SERVERS>',
    'properties.group.id' = '<CONSUMER_GROUP_ID>',
    'scan.startup.mode' = 'initial'
);

-- 2. 注册方舟图片向量化模型，并配置模型服务凭证
CREATE TEMPORARY MODEL tos_image_embedding
INPUT (payload STRING)
OUTPUT (embedding ARRAY<FLOAT>)
WITH (
    'provider' = 'ark',
    'endpoint' = 'https://ark.cn-beijing.volces.com/api/v3/embeddings/multimodal',
    'api-key' = '${secret_values.ark-api-key}',
    'model' = 'doubao-embedding-vision-251215',
    'model.dimensions' = '2048',
    'model.encoding-format' = 'float'
);

-- 3. 为私有 TOS 图片生成预签名地址，并封装为模型输入 JSON
CREATE TEMPORARY VIEW image_payload_input AS
SELECT
    object_key,
    object_url,
    bucket_name,
    file_name,
    object_etag,
    object_size,
    mtime,
    event_time,
    record_origin,
    IMAGE_URL_PAYLOAD(
        object_url,
        '${secret_values.tos-ak}',
        '${secret_values.tos-sk}',
        -- TOS 地域和 Endpoint 须与对象所在地域一致
        'cn-beijing',
        'tos-cn-beijing.volces.com',
        -- 预签名地址有效期应覆盖排队、重试和模型读取图片的时间
        CAST(3600 AS BIGINT)
    ) AS payload
FROM object_changes
-- 过滤对象标识或地址为空的记录
WHERE object_key IS NOT NULL AND object_url IS NOT NULL;

-- 4. 创建临时 print 表，保留对象信息和生成的向量
CREATE TEMPORARY TABLE embedding_output (
    object_key STRING NOT NULL,
    object_url STRING,
    bucket_name STRING,
    file_name STRING,
    object_etag STRING,
    object_size BIGINT,
    mtime TIMESTAMP_LTZ(3),
    event_time TIMESTAMP_LTZ(3),
    record_origin STRING,
    embedding ARRAY<FLOAT>
) WITH (
    'connector' = 'print'
);

-- 5. 执行推理，将对象信息和向量输出到 TaskManager 的 Stdout 日志
INSERT INTO embedding_output
SELECT
    object_key,
    object_url,
    bucket_name,
    file_name,
    object_etag,
    object_size,
    mtime,
    event_time,
    record_origin,
    embedding
FROM ML_PREDICT(
    TABLE image_payload_input,
    MODEL tos_image_embedding,
    -- 仅将 payload 列传给模型；其他对象字段在 SELECT 中保留
    DESCRIPTOR(payload)
);

示例中的 IMAGE_URL_PAYLOAD 为 TOS 对象生成预签名访问地址，并将地址封装为图片输入 JSON；模型服务通过该地址读取图片，模型调用发生在 ML_PREDICT。URL 有效期应覆盖作业排队、重试及模型读取媒体的时间。函数详情请参见IMAGE_URL_PAYLOAD。

查看结果 #

作业运行且模型请求成功后，可在 TaskManager 的 Stdout 日志中查看对象信息和对应向量。根据业务需要，可将输出表替换为下游系统对应的连接器，并保持向量字段类型与模型结果一致。将向量写入下游系统时，请确保向量索引配置的维度与模型输出的向量长度一致。

通过 http.connection-timeout、http.read-timeout 和 http.call-timeout 配置 HTTP 超时，通过 http.num-retry 配置重试次数。请求重试后仍失败时，按 retry-fallback-strategy 处理：

• FAILOVER：默认策略，向 Flink 传播调用错误。
• IGNORE：丢弃调用失败的输入记录并继续处理。因此，输出行数可能少于输入行数。

排查输出行数减少时，分别检查 NULL 输入和模型调用失败情况，并结合模型服务日志定位问题。

出现 404 或请求误入对话接口（Chat）。
检查 endpoint 的路径和服务实际开放的接口，并填写完整的多模态请求地址。多模态接口的识别依据是 URL 路径以 /embeddings/multimodal 结尾，仅在查询参数中出现该字符串无法识别为多模态接口。
模型调用返回 401 或 403。
检查 API Key、模型调用权限及 endpoint 所属服务。确保请求地址、API Key 和模型标识对应同一服务，并确认凭证具有模型调用权限。
文本推理成功，但图片或视频推理失败。
检查模型是否支持对应的输入类型、JSON 结构是否正确、媒体 URL 是否可由模型服务访问，以及预签名 URL 是否仍在有效期内。普通 URL 字符串会作为文本处理；图片和视频推理应使用对应的 JSON 结构。
模型调用返回 429、5xx 或出现超时。
检查服务配额、推理容量、请求耗时和作业并发。结合服务监控调整调用压力、超时与重试配置，并保留请求 ID（request ID）用于排查。
出现 JSON 解析错误或输入输出结构校验失败。
检查 JSON 中的 type、text、image_url.url 和 video_url.url 等字段，并核对模型 INPUT、OUTPUT 的字段数量及类型。按输入格式与输出说明修正 JSON，不支持的 type 会报错。
为什么输出行数少于输入行数？
分别检查用于推理的输入字段是否为 NULL，以及模型调用失败时是否最终采用 IGNORE 策略。NULL 输入会跳过推理，IGNORE 策略会丢弃失败行，因此输出行数可能少于输入行数。
写入下游系统时，向量维度不匹配。
使用 CARDINALITY(embedding) 检查模型实际输出的向量长度，并核对下游向量字段和索引配置。确保模型维度配置、实际输出维度与下游要求一致。
设置 model.dimensions 后，服务返回 does not support matryoshka representation。
移除 model.dimensions 参数后，使用模型默认维度重试。

Flink SQL AI 函数：查看图片输入构造、TOS 预签名、文件类型识别和内容读取等辅助函数的完整说明。