功能概述 #

VikingDB Flink SQL Connector 用于将火山引擎流式计算 Flink 版中的实时或批式计算结果写入 VikingDB Collection。Connector 根据稳定主键执行 Upsert 或 Delete，并支持通过 VikingDB Catalog 查看 Collection 元数据以及创建、删除 Collection。
Connector 标识符和 Catalog 类型均为 vikingdb。Connector 已由平台内置，无需上传额外 Connector JAR。

版本兼容性 #

特色功能 #

• 支持将 Flink Changelog 转换为 VikingDB Upsert 和 Delete 操作，用于实时同步业务数据和向量数据。
• 支持通过 ARRAY<FLOAT>、ARRAY<DOUBLE>、二维数组和稀疏 Map 写入稠密向量、Tensor 和稀疏向量。
• 支持写入启用文本、图片或视频向量化能力的 Collection。
• 支持使用 Catalog 查看 Collection 元数据，并通过 Flink SQL 创建或删除真实 Collection。
• API Key、AK 和 SK 均支持通过 Flink 加密变量或运行环境变量引用。

前提条件 #

1. 已开通火山引擎流式计算 Flink 版和向量数据库 VikingDB。
2. Flink 资源池可访问 VikingDB 数据面地址；生产环境建议优先采用私网连接，具体操作参见私网连接方式。
3. 使用 Sink 时，已创建目标 Collection，并确认字段名、字段类型、主键和向量维度与 Flink Sink 表一致。
4. 已获取 VikingDB API Key，并通过 Flink 加密变量或运行环境变量提供给作业。
5. 使用 Catalog 时，已获取控制面 AK、SK 和数据面 API Key，并确认凭证具有所需的 Collection 管理权限。

使用限制 #

• Connector 不提供 Source 和 Lookup 能力。
• Sink 必须且只能配置一个主键字段，主键值不能为 NULL。
• sink.bulk-flush.batch-size 取值范围为 1～100，默认值为 100。
• 向量字段必须与 Collection 的维度一致。通过 ARRAY<FLOAT> 或 ARRAY<DOUBLE> 推断 vector 时，需要配置 vikingdb.field.<字段名>.dim。
• BYTES、VARBINARY、ARRAY<BOOLEAN> 等不支持映射的 Flink 类型会在建表或运行阶段报错。
• 外部写入采用 At-Least-Once 语义。故障恢复期间可能重放数据，应通过稳定主键和 VikingDB Upsert/Delete 使最终结果收敛。
• async=true 可能带来分钟级 Collection 可见延迟和小时级 Index 同步延迟，不适用于要求写入后立即检索的场景。
• 写入启用向量化的 Collection 时，应写入 text、image 或 video 等原始字段；写入已有向量的 Collection 时，应写入 vector 字段，避免同时提交互斥字段。
• Catalog 不支持重命名或修改 Collection，也不支持分区、视图和函数。
• Catalog CREATE TABLE 和 DROP TABLE 会直接创建或删除 VikingDB 中的真实 Collection，不是只修改 Flink Catalog 元数据。

语法结构 #

CREATE TABLE vikingdb_sink (
    id STRING,
    text STRING,
    vector ARRAY<DOUBLE>,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'vikingdb',
    'data-plane.host' = '<vikingdb_data_plane_host>',
    'collection-name' = '<collection_name>',
    'api-key' = '${secret_values.vikingdb-api-key}',
    'vikingdb.field.vector.dim' = '768'
);

参数说明 #

通用连接参数 #

Sink 独有参数 #

Catalog 独有参数 #

Catalog 同时使用“通用连接参数”中的数据面地址、地域和 API Key。

类型映射 #

使用示例 #

写入 Collection #

以下示例将 Flink SQL 结果写入已存在的 VikingDB Collection：

CREATE TABLE vikingdb_sink (
    id STRING,
    text STRING,
    city STRING,
    score DOUBLE,
    vector ARRAY<DOUBLE>,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'vikingdb',
    'data-plane.host' = '<vikingdb_data_plane_host>',
    'data-plane.scheme' = 'https',
    'region' = '<region>',
    'collection-name' = '<collection_name>',
    'api-key' = '${secret_values.vikingdb-api-key}',
    'vikingdb.field.vector.dim' = '768',
    'sink.bulk-flush.batch-size' = '100',
    'sink.bulk-flush.interval' = '1s',
    'sink.local-time-zone' = 'UTC'
);

INSERT INTO vikingdb_sink
SELECT id, text, city, score, vector
FROM source_table;

也可以通过 'api-key-env' = 'VIKINGDB_API_KEY' 引用环境变量。不要同时配置 api-key 和 api-key-env。
预期结果。 INSERT 和 UPDATE_AFTER 按主键 Upsert；DELETE 按主键删除；默认情况下 UPDATE_BEFORE 按旧主键执行 Delete。

查看 Collection 元数据 #

CREATE CATALOG viking_catalog WITH (
    'type' = 'vikingdb',
    'control-plane.host' = '<vikingdb_control_plane_host>',
    'data-plane.host' = '<vikingdb_data_plane_host>',
    'data-plane.scheme' = 'https',
    'region' = '<region>',
    'project-name' = 'default',
    'default-database' = 'default',
    'access-key' = '${secret_values.vikingdb-access-key}',
    'secret-key' = '${secret_values.vikingdb-secret-key}',
    'api-key' = '${secret_values.vikingdb-api-key}'
);

USE CATALOG viking_catalog;
USE `default`;

SHOW TABLES;
DESCRIBE <collection_name>;

AK、SK 和 API Key 也可以分别通过 access-key-env、secret-key-env 和 api-key-env 引用运行环境变量。

运行机制 #

Connector 采用 At-Least-Once 语义。Checkpoint 恢复期间可能重放请求，稳定、唯一且非空的主键是最终结果收敛的前提。
Collection 数据写入成功后，Index 仍需要按照 VikingDB 的同步机制更新检索数据。启用 async=true 或使用异步向量化时，可检索时间可能明显晚于 Collection 点查时间。

验证与排障 #

验证方法 #

1. 提交作业并确认已进入 RUNNING，日志中没有字段类型、向量维度、凭证或连接错误。
2. 写入一条新主键记录，通过 Collection 点查核对字段内容。
3. 更新相同主键，确认记录被覆盖，而不是新增重复数据。
4. 产生 Delete 事件，确认 Collection 中的对应记录被删除。
5. 检查向量数组长度与 Collection 的 dim 一致，并通过目标 Index 执行样例检索。
6. 等待 Checkpoint 成功后执行受控重启，确认相同主键的最终记录仍然正确。

常见问题 #

注意事项 #

• Catalog CREATE TABLE 和 DROP TABLE 会直接影响 VikingDB 中的真实 Collection。尤其是 DROP TABLE 会永久删除 Collection 及全部数据，必须谨慎操作。
• 生产环境必须通过 Flink 加密变量或运行环境变量提供 API Key、AK 和 SK，不要在 SQL 中写入明文凭证。

相关文档 #

• VikingDB：数据写入 UpsertData
• VikingDB：创建数据集
• VikingDB：更新数据集
• VikingDB：删除数据集
• 流式计算 Flink 版：上线 Flink 任务