AI 函数将模型推理融入火山引擎流式计算 Flink 版的 SQL 数据处理流程。您可以在 SQL 作业中接入模型服务,对输入数据进行智能处理,并将推理结果与过滤、关联等 SQL 操作结合,完成后续业务计算。
AI SQL 围绕模型的使用提供以下能力:
当前版本支持多模态 Embedding 服务接入及配套的数据准备函数。您可以将文本、图片等数据转换为向量,为下游语义检索和多模态检索准备数据。实际支持的输入类型取决于所选服务和模型。
本文介绍的能力要求 Flink 2.2-volcano 及以上引擎版本;各函数的其他限制见对应章节。
模型定义、推理调用和完整示例见Flink SQL 多模态 Embedding 使用指南。下文介绍配套数据准备函数的语法、参数、使用限制及示例。
函数 | 语法 | 返回类型和说明 |
|---|---|---|
| STRING,为 TOS 对象生成临时 GET 访问 URL。 | |
| STRING,构造图片 URL 的 JSON 输入;支持先对私有 TOS 对象预签名。 | |
| STRING,根据文件扩展名获取 MIME 类型。 | |
| BYTES,异步读取 TOS 对象的完整二进制内容。 |
PRESIGN_URL 根据火山引擎对象存储 TOS(Torch Object Storage)的对象地址和访问凭证,生成带签名的临时 HTTPS 访问地址。您可以将该地址提供给模型服务,使其在有效期内读取私有图片等文件。
tos://<bucket>/<object-key>,bucket 和 object-key 均不能为空。?...)、片段标识(fragment,#...)、用户名(如 user@)或端口(如 :8443)。对象 key 中的保留字符需使用百分号编码,例如对象名 cat#1.jpg 应写为 cat%231.jpg。此限制仅针对输入对象地址。PRESIGN_URL(uri, access_key, secret_key) PRESIGN_URL(uri, access_key, secret_key, region, endpoint) PRESIGN_URL(uri, access_key, secret_key, region, endpoint, expires_seconds)
参数 | 类型 | 说明 |
|---|---|---|
| STRING | TOS 对象 URI |
| STRING | AK,不能为空字符串 |
| STRING | SK,不能为空字符串 |
| STRING | 默认 |
| STRING | 默认 |
| BIGINT | 有效期,单位秒,默认 3600,须大于 0 |
返回包含 TOS 签名参数的 GET URL。任一参数为 NULL 时返回 NULL。
URI 无效、必填参数为空字符串、有效期非正数或签名失败时会报错,不会自动返回 NULL。
|
| 地域 / Endpoint | 有效期(秒) |
|---|---|---|---|
default-options |
| 默认北京配置 | 3600 |
custom-endpoint |
|
| 3600 |
custom-expiry |
|
| 180 |
使用具有对应对象读取权限的 AK/SK,并将示例中的桶、地域和 Endpoint 替换为相互匹配的配置。
CREATE TEMPORARY VIEW signed_images AS SELECT 'default-options' AS case_name, PRESIGN_URL( 'tos://<BEIJING_BUCKET>/images/demo.jpg', '${secret_values.tos-ak}', '${secret_values.tos-sk}' ) AS signed_url UNION ALL SELECT 'custom-endpoint', PRESIGN_URL( 'tos://<SHANGHAI_BUCKET>/images/demo.jpg', '${secret_values.tos-ak}', '${secret_values.tos-sk}', 'cn-shanghai', 'tos-cn-shanghai.volces.com' ) UNION ALL SELECT 'custom-expiry', PRESIGN_URL( 'tos://<SHANGHAI_BUCKET>/images/demo.jpg', '${secret_values.tos-ak}', '${secret_values.tos-sk}', 'cn-shanghai', 'tos-cn-shanghai.volces.com', CAST(180 AS BIGINT) ); CREATE TEMPORARY TABLE presign_result ( case_name STRING, url_not_null BOOLEAN, uses_https BOOLEAN, has_signature BOOLEAN ) WITH ('connector' = 'print'); -- 验证输出结构,避免将完整签名 URL 打印到日志。 INSERT INTO presign_result SELECT case_name, signed_url IS NOT NULL AS url_not_null, signed_url LIKE 'https://%' AS uses_https, signed_url LIKE '%X-Tos-Signature=%' AS has_signature FROM signed_images;
替换有效配置并成功执行后,presign_result 结果表通过 print 连接器输出到 TaskManager 日志。下表省略日志前缀,行顺序以实际输出为准。
|
|
|
|
|---|---|---|---|
default-options | true | true | true |
custom-endpoint | true | true | true |
custom-expiry | true | true | true |
signed_url 是包含 X-Tos-Algorithm、X-Tos-Credential、X-Tos-Expires、X-Tos-Date 和 X-Tos-Signature 等查询参数的 GET URL。三个用例的有效期依次为 3600、3600、180 秒。签名值随凭证、对象地址及调用时间变化,不是固定字符串。
IMAGE_URL_PAYLOAD 将图片地址封装为含 image_url 元素的 JSON 数组字符串,可作为 ML_PREDICT 的输入。处理火山引擎对象存储 TOS 中的私有图片时,带密钥形式会先生成预签名访问地址,再完成封装。
tos://<bucket>/<object-key> 格式,bucket 和 object-key 均不能为空。对象地址不允许附带查询参数(query,?...)、片段标识(fragment,#...)、用户名(如 user@)或端口(如 :8443)。访问凭证通过 access_key 和 secret_key 参数提供,服务地址通过 endpoint 参数指定。对象 key 中的保留字符需使用百分号编码,例如对象名 cat#1.jpg 应写为 cat%231.jpg。IMAGE_URL_PAYLOAD(uri) IMAGE_URL_PAYLOAD(uri, access_key, secret_key) IMAGE_URL_PAYLOAD(uri, access_key, secret_key, region, endpoint, expires_seconds)
参数 | 类型 | 说明 |
|---|---|---|
| STRING | 单参数时原样封装图片 URI/URL;带密钥时为待签名的 TOS URI |
| STRING | TOS AK,不能为空字符串 |
| STRING | TOS SK,不能为空字符串 |
| STRING | 默认 |
| STRING | 默认 |
| BIGINT | 有效期,单位秒,默认 3600,须大于 0 |
显式指定地域和 Endpoint 时,必须同时提供 expires_seconds。
返回含一个 image_url 元素的 JSON 数组字符串:
[{"type":"image_url","image_url":{"url":"<url>"}}]
任一实际传入的参数为 NULL 时返回 NULL。单参数形式保留原始 URI;带密钥形式先生成预签名 URL,再写入 image_url.url。
带密钥形式遇到无效 TOS URI、空签名参数、非正有效期或签名失败时会报错。
|
|
|---|---|
1 |
|
2 |
|
CREATE TEMPORARY VIEW images(id, uri) AS VALUES (1, 'https://example.com/images/cat.jpg'), (2, CAST(NULL AS STRING)); CREATE TEMPORARY TABLE image_url_result ( id INT, image_payload STRING ) WITH ('connector' = 'print'); INSERT INTO image_url_result SELECT id, IMAGE_URL_PAYLOAD(uri) AS image_payload FROM images;
下表按 id 展示预期结果,实际输出顺序可能不同。
|
|
|---|---|
1 |
|
2 |
|
image_url_result 结果表通过 print 连接器输出到 TaskManager 日志,上表省略日志前缀。这个示例只构造 JSON,不访问图片地址。实际调用模型时,请将示例 URL 替换为模型服务可访问的真实图片地址。
调用形式 | 对象 URI | 地域 | Endpoint | 有效期(秒) |
|---|---|---|---|---|
3 参数 |
|
|
| 3600 |
6 参数 |
|
|
| 600 |
示例中的对象地址和凭证为占位符,请替换为实际配置。建议通过平台的密钥管理能力注入凭证,避免在日志中记录密钥。该示例会将包含预签名 URL 的 payload 输出到日志,仅用于开发验证,请避免长期保留或共享日志。
-- 使用默认北京配置和默认 3600 秒有效期。 CREATE TEMPORARY TABLE tos_image_url_default_result ( case_name STRING, image_payload STRING ) WITH ('connector' = 'print'); INSERT INTO tos_image_url_default_result SELECT 'default-options' AS case_name, IMAGE_URL_PAYLOAD( 'tos://<BUCKET>/images/cat.png', '${secret_values.tos-ak}', '${secret_values.tos-sk}' ) AS image_payload;
-- 显式指定地域、Endpoint 和 600 秒有效期。 CREATE TEMPORARY TABLE tos_image_url_custom_result ( case_name STRING, image_payload STRING ) WITH ('connector' = 'print'); INSERT INTO tos_image_url_custom_result SELECT 'custom-expiry' AS case_name, IMAGE_URL_PAYLOAD( 'tos://<BUCKET>/images/cat.png', '${secret_values.tos-ak}', '${secret_values.tos-sk}', 'cn-beijing', 'tos-cn-beijing.volces.com', CAST(600 AS BIGINT) ) AS image_payload;
以上两个示例分别执行后,预期在各自任务的 TaskManager Stdout 日志中看到对应结果。下表省略 print 连接器的日志前缀;占位符代表实际生成的预签名 URL,两次调用的有效期分别为 3600 秒和 600 秒。
|
|
|---|---|
|
|
|
|
| 对象 URI |
|---|---|
1 |
|
先准备一张真实图片、可读取该对象的 TOS AK/SK,以及有权限调用的火山方舟大模型服务平台(Ark)的多模态 Embedding 模型及 API Key。模型服务需能访问签名 URL。
模型定义、推理调用相关内容请参见Flink SQL 多模态 Embedding 使用指南。
替换模型标识、凭证及对象地址,并配置所需模型服务依赖后执行。
SET 'execution.runtime-mode' = 'streaming'; CREATE TEMPORARY TABLE image_events ( id BIGINT ) WITH ( 'connector' = 'datagen', 'number-of-rows' = '1', 'fields.id.kind' = 'sequence', 'fields.id.start' = '1', 'fields.id.end' = '1' ); CREATE TEMPORARY VIEW image_input AS SELECT id, IMAGE_URL_PAYLOAD( 'tos://<BUCKET>/images/cat.png', '${secret_values.tos-ak}', '${secret_values.tos-sk}', 'cn-beijing', 'tos-cn-beijing.volces.com', CAST(3600 AS BIGINT) ) AS image_payload FROM image_events; CREATE TEMPORARY MODEL image_embedding_model INPUT (image_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' = '<MODEL_OR_ENDPOINT_ID>' ); CREATE TEMPORARY TABLE embedding_output ( id BIGINT, image_payload STRING, embedding ARRAY<FLOAT> ) WITH ('connector' = 'print'); INSERT INTO embedding_output SELECT id, image_payload, embedding FROM ML_PREDICT( TABLE image_input, MODEL image_embedding_model, DESCRIPTOR(image_payload) );
成功执行后,预期在 TaskManager 的 Stdout 日志中看到图片输入 image_payload 和模型返回的 embedding。下表省略 print 连接器的日志前缀,URL 为占位符。
|
|
|
|---|---|---|
1 |
| 所选模型返回的 |
向量数值和维度由图片内容及模型配置决定。该链路由模型服务通过 URL 获取图片,无需先调用 FETCH_CONTENT。
MIME_TYPE 根据文件地址或路径的扩展名返回 MIME(Multipurpose Internet Mail Extensions,多用途互联网邮件扩展)类型,即以“主类型/子类型”形式标识文件内容类型的字符串,例如 image/jpeg。您可以据此按扩展名区分图片、PDF 等文件,选择后续处理方式。
? 后的查询参数(query)和 # 后的片段标识(fragment)不参与文件类型判断。例如 https://example.com/cat.jpg?download=1#preview 仍按 .jpg 判断,返回 image/jpeg。MIME_TYPE(uri)
uri 为 STRING,可包含查询参数(?...)和片段标识(#...);这两部分不参与文件类型判断。
返回 STRING。输入为 NULL 时返回 NULL;未知或无扩展名时返回 application/other。
扩展名 | MIME 类型 |
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
其他或无扩展名 |
|
|
|
|---|---|
1 |
|
2 |
|
3 |
|
4 |
|
5 |
|
6 |
|
7 |
|
8 |
|
9 |
|
10 |
|
11 |
|
12 |
|
示例可直接执行,不需要配置 TOS 凭证或准备真实文件。
CREATE TEMPORARY VIEW files(id, uri) AS VALUES (1, 'tos://demo-bucket/image.bmp'), (2, 'tos://demo-bucket/image.jpg'), (3, 'tos://demo-bucket/image.JPEG'), (4, 'tos://demo-bucket/image.png'), (5, 'tos://demo-bucket/image.tiff'), (6, 'tos://demo-bucket/image.webp'), (7, 'tos://demo-bucket/image.heic'), (8, 'tos://demo-bucket/document.pdf'), (9, 'tos://demo-bucket/archive.zip'), (10, 'tos://demo-bucket/no-extension'), (11, CAST(NULL AS STRING)), (12, 'https://example.com/image.JPG?download=1#preview'); CREATE TEMPORARY TABLE mime_type_result ( id INT, uri STRING, mime_type STRING ) WITH ('connector' = 'print'); INSERT INTO mime_type_result SELECT id, uri, MIME_TYPE(uri) AS mime_type FROM files;
成功执行后,预期在 TaskManager 的 Stdout 日志中看到以下结果。下表省略 print 连接器的日志前缀,并按 id 展示;实际输出顺序可能不同。
|
|
|
|---|---|---|
1 |
|
|
2 |
|
|
3 |
|
|
4 |
|
|
5 |
|
|
6 |
|
|
7 |
|
|
8 |
|
|
9 |
|
|
10 |
|
|
11 |
|
|
12 |
|
|
FETCH_CONTENT 在流模式下异步读取火山引擎对象存储 TOS 中指定对象的完整内容,返回 BYTES 类型的二进制数据。适用于在 SQL 中直接获取文件内容,并将结果传递给下游处理的场景。
tos:// 对象地址,不支持下载 HTTP/HTTPS URL。tos://<bucket>/<object-key> 格式,bucket 和 object-key 均不能为空。对象地址不允许附带查询参数(query,?...)、片段标识(fragment,#...)、用户名(如 user@)或端口(如 :8443)。对象 key 中的保留字符需使用百分号编码,例如对象名 cat#1.jpg 应写为 cat%231.jpg。FETCH_CONTENT(uri) FETCH_CONTENT(uri, concurrency)
参数 | 类型 | 说明 |
|---|---|---|
| STRING | TOS 对象 URI |
| INT | 可选,默认 |
并发按函数实例生效,在同一实例内固定。
返回 BYTES类型的对象完整二进制内容。uri 为 NULL 时返回 NULL。
对象不存在、目标为目录、文件超限、下载超时或文件系统读取失败时计算失败,不会自动返回 NULL。对象被覆盖后,相同 URI 可能读到不同内容。
以下配置用于控制下载大小、缓冲区、并发和超时。
配置 | 默认值 | 说明 |
|---|---|---|
| 10 MiB | 单个对象大小上限,须大于 0 且小于 2 GiB;超限失败,不截断 |
| 64 KiB | 须大于 0;有效缓冲区至少 64 KiB,但不超过内容大小上限 |
| 1 分钟 | 单次下载读取超时,覆盖解析、元数据查询、打开和读取 |
| 10 | 异步标量函数框架并发上限 |
| 3 分钟 | 异步结果等待超时,可能包含排队开销 |
说明
读取超时从下载任务开始执行时计时,异步等待还可能包含排队开销。两者分别生效,仅增大异步等待超时不会延长下载读取超时。
先将以下两个 UTF-8 文本文件上传到对应 TOS 路径。文件内容不包含额外换行;第三条输入用于展示 NULL 行为。
|
| 文件内容 |
|---|---|---|
1 |
| Hello, Flink! |
2 |
| Streaming AI |
3 |
| 不读取文件 |
配置 TOS 文件系统访问凭证后,替换为实际对象路径并运行。以下语句使用下载并发 2;省略第二个参数时下载并发为 1。
CREATE TEMPORARY TABLE file_events ( id BIGINT ) WITH ( 'connector' = 'datagen', 'number-of-rows' = '3', 'fields.id.kind' = 'sequence', 'fields.id.start' = '1', 'fields.id.end' = '3' ); CREATE TEMPORARY VIEW files AS SELECT id, CASE id WHEN 1 THEN 'tos://demo-bucket/documents/hello.txt' WHEN 2 THEN 'tos://demo-bucket/documents/ai.txt' ELSE CAST(NULL AS STRING) END AS uri FROM file_events; CREATE TEMPORARY TABLE content_output ( id BIGINT, uri STRING, content STRING ) WITH ( 'connector' = 'print' ); INSERT INTO content_output SELECT id, uri, DECODE(FETCH_CONTENT(uri, 2), 'UTF-8') AS content FROM files;
在上述文件内容和访问配置下,content_output 结果表通过 print 连接器输出到 TaskManager 日志。下表省略日志前缀,并按 id 展示预期结果。
|
|
|
|---|---|---|
1 |
| Hello, Flink! |
2 |
| Streaming AI |
3 |
|
|
DECODE(..., 'UTF-8') 仅适用于已知为 UTF-8 的文本文件。对于图片、PDF 等文件,请直接将 FETCH_CONTENT 返回的二进制数据交给支持对应格式的下游处理,不要使用 DECODE(..., 'UTF-8') 将其转换为文本。