文档中心

流式计算 Flink版

CDC Connector 参考

VikingDB （公测中）

功能概述

VikingDB CDC Pipeline Sink Connector 用于将火山引擎流式计算 Flink 版 Pipeline 作业捕获的全量数据和增量变更持续写入 VikingDB Collection，适用于实时检索、召回和 RAG 数据链路。

类别	详情
支持类型	Sink
作业类型	Flink CDC Pipeline YAML
运行模式	流模式
支持的变更类型	INSERT、REPLACE、UPDATE、DELETE；DDL 支持 CREATE TABLE、DROP TABLE、TRUNCATE TABLE，不支持 Schema Evolution
数据格式	VikingDB Collection Record
组件形式	平台内置，无需上传额外 Connector JAR
交付语义	At-Least-Once；通过稳定主键的幂等 Upsert/Delete 使最终结果收敛

版本兼容性

组件	支持版本	说明
火山引擎流式计算 Flink 版	1.16.0 1.20.0	Connector 已由平台内置。
Flink CDC	3.4及以上版本	创建 Pipeline 作业时选择 3.4 或以上版本。
VikingDB	火山引擎向量数据库 VikingDB	目标 Collection 的字段、主键和向量维度必须与 CDC Schema 兼容。

特色功能

支持将 Insert、Replace 和 Update 转换为 Upsert，将 Delete 转换为按主键删除。
支持根据 CDC CREATE TABLE 事件创建或校验 VikingDB Collection。
支持根据源表标识生成 Collection 名称，或使用模板自定义多表路由。
支持字段类型覆盖、稠密向量、稀疏向量和列表字段转换。
支持批量写入和可选异步 Upsert。

前提条件

已开通火山引擎流式计算 Flink 版和向量数据库 VikingDB。
Flink 资源池能够访问 VikingDB 数据面和控制面地址。
上游表已定义单列主键，且主键字段类型和数据满足本 Connector 限制。
已准备 VikingDB API Key；需要管理 Collection 时，还需准备具有相应权限的火山引擎 AK/SK。
已在 Flink 中创建加密变量，不在 Pipeline YAML 中保存明文凭证。
写入已有 Collection 时，已确认字段名称、类型、主键和向量维度与上游 CDC Schema 兼容。

使用限制

Connector 仅支持作为 Flink CDC Pipeline Sink。
每张源表必须且只能有一个主键，主键值不能为 NULL。
主键仅支持 CHAR、VARCHAR、TINYINT、SMALLINT、INTEGER、BIGINT 和 scale 为 0 的 DECIMAL。
当 Update 前后主键不同且未忽略 Update Before 时，Sink 并行度必须为 1；并行度大于 1 会拒绝该类事件。
sink.bulk-flush.batch-size 的取值范围为 1～100。
默认不忽略 Delete。启用 sink.ignore-delete 会使目标端保留源端已删除的记录。
当前仅支持 Create Table、Drop Table 和 Truncate Table 三类 DDL，不支持 Add Column、Drop Column、Rename Column 和 Alter Type 等 Schema Evolution；收到不支持的 Schema 变更时作业会报错。
VikingDB 写入采用 At-Least-Once 语义。故障恢复期间可能重放请求，必须使用稳定、唯一且非空的主键，使幂等 Upsert/Delete 的最终结果收敛。
向量字段的实际维度必须与 Collection Schema 一致。
异步写入或向量化可能使 Collection 写入成功时间和 Index 可检索时间存在差异。

语法结构

source:
  type: <source_type>
  # Source 参数

sink:
  type: vikingdb
  region: <region>
  control-plane.host: <vikingdb_control_plane_host>
  data-plane.host: <vikingdb_data_plane_host>
  api-key: ${secret_values.vikingdb-api-key}
  access-key: ${secret_values.vikingdb-access-key}
  secret-key: ${secret_values.vikingdb-secret-key}
  collection-name-template: "${database}_${table}"
  auto-create-table: true

pipeline:
  name: <pipeline_name>
  parallelism: 1
  local-time-zone: Asia/Shanghai

参数说明

连接与鉴权参数

参数	说明	数据类型	是否必填	默认值	备注
`type`	Pipeline Sink 类型。	STRING	是	无	固定为 `vikingdb`。
`region`	VikingDB 地域。	STRING	是	无	例如 `cn-beijing`；关闭自动建表时仍需配置。
`control-plane.host`	VikingDB 控制面地址。	STRING	是	无	关闭自动建表时仍需配置。
`data-plane.host`	VikingDB 数据面地址。	STRING	是	无	Upsert、Delete 和清空数据时使用。
`data-plane.scheme`	数据面访问协议。	STRING	否	`https`	支持 `http`、`https`；生产环境建议使用 HTTPS。
`api-key`	VikingDB 数据面 API Key。	STRING	条件必填	无	与 `api-key-env` 选择一种；建议引用 Flink 加密变量。
`api-key-env`	保存 API Key 的环境变量名。	STRING	条件必填	无	与 `api-key` 选择一种。
`access-key`	控制面使用的火山引擎 Access Key。	STRING	条件必填	无	始终需要；与 `access-key-env` 选择一种，建议引用 Flink 加密变量。
`access-key-env`	保存 Access Key 的环境变量名。	STRING	条件必填	无	与 `access-key` 选择一种。
`secret-key`	控制面使用的火山引擎 Secret Key。	STRING	条件必填	无	始终需要；与 `secret-key-env` 选择一种，建议引用 Flink 加密变量。
`secret-key-env`	保存 Secret Key 的环境变量名。	STRING	条件必填	无	与 `secret-key` 选择一种。

Collection 与写入参数

参数	说明	数据类型	是否必填	默认值	备注
`collection-name-template`	根据源表生成 Collection 名称的模板。	STRING	否	源表标识，点号替换为下划线	支持 `${database}`、`${schema}`、`${table}`。
`resource-id`	VikingDB Collection 资源 ID。	STRING	否	无	用于数据面写入。
`project-name`	VikingDB Project 名称。	STRING	否	`default`	控制面操作使用。
`default-database`	控制面配置中的逻辑 Catalog Database。	STRING	否	`default`	不代表实际创建独立数据库。
`auto-create-table`	收到 Create Table 事件时是否自动创建 Collection。	BOOLEAN	否	`true`	关闭后仍会拒绝不支持的 Schema 事件。
`ttl`	写入记录的生存时间。	LONG	否	无	单位为秒；配置值必须大于 0。
`async`	是否使用异步 Upsert。	BOOLEAN	否	`false`	启用前评估数据可见延迟。
`sink.bulk-flush.batch-size`	单批最多写入的记录数。	INTEGER	否	`100`	取值范围为 1～100。
`sink.bulk-flush.interval`	未满批次的最长刷新间隔。	DURATION	否	`1s`	`0` 表示关闭按时间触发刷新。
`sink.ignore-update-before`	是否忽略 Update Before。	BOOLEAN	否	`false`	开启后，主键变化的 Update 不删除旧主键记录。
`sink.ignore-delete`	是否忽略 Delete。	BOOLEAN	否	`false`	开启后目标端保留源端已删除记录。

连接参数

参数	说明	数据类型	是否必填	默认值	备注
`connection.timeout`	建立连接的超时时间。	DURATION	否	`10s`	必须大于 0。
`socket.timeout`	数据面 Socket 超时时间。	DURATION	否	`120s`	必须大于 0。
`read.timeout`	控制面读取超时时间。	DURATION	否	`120s`	必须大于 0。
`max-retries`	请求失败后的最大重试次数。	INTEGER	否	`3`	必须大于等于 0。
`connection.max-total`	HTTP 连接池最大连接数。	INTEGER	否	`200`	必须大于 0。
`connection.max-per-route`	单 Route 最大连接数。	INTEGER	否	`200`	必须大于 0。
`connection.time-to-live`	HTTP 连接的最长存活时间。	DURATION	否	`60s`	必须大于等于 0。
`connection.validate-after-inactivity`	空闲连接复用前的校验阈值。	DURATION	否	`10s`	必须大于等于 0。
`connection.idle-evict`	空闲连接清理间隔。	DURATION	否	`30s`	必须大于等于 0。

字段级参数

字段级参数可在 Sink 中配置，也可通过 Schema Options 传入；Schema Options 中的同名配置优先。

参数	说明	数据类型	是否必填	默认值	备注
`vikingdb.field.<field_name>.type`	覆盖字段的 VikingDB 类型。	STRING	否	自动推断	可选值包括 `vector`、`sparse_vector`、`list_string`、`list_int64`、`date_time`、`video`、`image`、`geo_point`、`text`。
`vikingdb.field.<field_name>.dim`	指定向量字段维度。	INTEGER	条件必填	无	自动创建向量字段时通常需要，且必须与实际向量长度一致。
`vikingdb.vectorize.<param>`	配置自动建表时使用的向量化参数。	STRING	否	无	正式开放参数；自动建表时透传给 VikingDB 控制面。

类型映射

Flink CDC 类型	VikingDB 类型或写入形态	说明
`CHAR`、`VARCHAR`	`string`	可通过字段级参数覆盖为部分特殊类型。
`BOOLEAN`	`bool`	写入布尔值。
`TINYINT`、`SMALLINT`、`INTEGER`、`BIGINT`	`int64`	整数类型可作为主键。
`DECIMAL(p, 0)`	`int64`	可作为主键。
`DECIMAL(p, s)`，`s > 0`	`float`	不支持作为主键，使用前评估精度变化。
`FLOAT`、`DOUBLE`	`float`	向量元素必须为有限数值。
`DATE`、`TIME`、`TIMESTAMP`、`TIMESTAMP_LTZ`、`TIMESTAMP_TZ`	`date_time` 字符串	按 Pipeline `local-time-zone` 或值自身偏移生成 ISO 8601 字符串。
`ARRAY`	`vector` 或 `list`	向量需要配置正确维度；也可显式指定列表类型。
`MAP`	`map` 或 `sparse_vector`	稀疏向量的 value 会转换为 Double。
`ROW`	`map`	显式配置为 `video` 时可转换为视频对象。

使用示例

MySQL 数据同步到 VikingDB

以下示例将 MySQL 表 <source_database>.<source_table> 的存量数据和增量变更同步到 VikingDB。运行前请替换地址、表名、地域和加密变量名称。

source:
  type: mysql
  name: MySQL Source
  hostname: <mysql_host>
  port: 3306
  username: ${secret_values.mysql-username}
  password: ${secret_values.mysql-password}
  tables: <source_database>.<source_table>
  server-id: 5401-5404

sink:
  type: vikingdb
  name: VikingDB Sink
  region: <region>
  control-plane.host: <vikingdb_control_plane_host>
  data-plane.host: <vikingdb_data_plane_host>
  api-key: ${secret_values.vikingdb-api-key}
  access-key: ${secret_values.vikingdb-access-key}
  secret-key: ${secret_values.vikingdb-secret-key}
  project-name: default
  default-database: default
  collection-name-template: "${database}_${table}"
  auto-create-table: true
  sink.bulk-flush.batch-size: 100
  sink.bulk-flush.interval: 1s
  vikingdb.field.embedding.type: vector
  vikingdb.field.embedding.dim: 1024
  vikingdb.field.tags.type: list_string

pipeline:
  name: MySQL to VikingDB Pipeline
  parallelism: 1
  local-time-zone: Asia/Shanghai

预期结果： 全量阶段创建或校验目标 Collection 并写入已有记录；增量阶段按主键 Upsert 新增和更新记录，并默认删除源端已删除的记录。

运行机制

CDC 事件	VikingDB 操作	说明
`INSERT`、`REPLACE`	Upsert	按主键写入完整记录。
`UPDATE`	Upsert；主键变化时先删除旧主键	主键变化且未忽略 Update Before 时，并行度必须为 1。
`DELETE`	Delete	`sink.ignore-delete: true` 时忽略。
`CREATE TABLE`	创建或校验 Collection	开启自动建表时创建不存在的 Collection；已存在时执行兼容性校验。
`DROP TABLE`	删除 Collection	会永久删除真实 Collection 及其中数据。
`TRUNCATE TABLE`	清空 Collection	会删除 Collection 中的全部记录。
其他 Schema 变更	报错	暂不支持 Schema Evolution。

Connector 写入 VikingDB 的交付语义为 At-Least-Once。Checkpoint 用于保存 Pipeline 的处理进度；故障恢复后可能重新发送尚未完成确认的数据。稳定、唯一且非空的主键可使重复的 Upsert/Delete 幂等执行，让目标端最终结果收敛到与 Exactly-Once 等效的业务结果。

验证与排障

验证方法

提交作业并确认进入运行状态，日志中没有连接、鉴权、字段类型或向量维度错误。
等待全量阶段完成，通过 VikingDB Collection 点查核对记录主键和字段内容。
新增、更新、删除同一测试主键，分别验证 Upsert 覆盖和 Delete 结果。
如果存在主键变化 Update，使用并行度 1 验证旧主键已删除、新主键已写入。
核对向量数组长度与 Collection Schema 中的维度一致，并在关联 Index 中执行样例检索。
等待成功 Checkpoint 后执行受控重启，确认相同主键的最终记录没有出现错误回退。

常见问题

现象	可能原因	处理建议
提示源表主键不受支持	未定义主键、定义了复合主键、主键为空或主键类型不支持。	将源表调整为单列非空主键，并使用受支持的字符串或整数类型。
向量字段自动建表失败	未指定向量类型或维度。	配置 `vikingdb.field.<field>.type: vector` 和正确的 `dim`。
已存在 Collection 校验失败	字段名、类型、主键或向量维度与 CDC Schema 不一致。	修改 Transform/字段配置，或准备兼容的目标 Collection。
主键变化 Update 被拒绝	Sink 并行度大于 1。	将 `pipeline.parallelism` 设置为 1，或避免修改主键。
源端删除后目标记录仍存在	启用了 `sink.ignore-delete`，或 Delete 未到达 Sink。	检查 Sink 配置、源端 CDC 日志和 Pipeline 事件。
Collection 已写入但暂时无法检索	Index 尚未同步，或启用了异步写入/向量化。	先通过 Collection 点查确认，再检查 Index 同步状态。

注意事项

DROP TABLE 会删除真实 VikingDB Collection 及其中全部数据；删除前必须先处理关联 Index，并完成目标核对、备份和变更审批。
TRUNCATE TABLE 会清空目标 Collection 中的全部记录。仅在已确认影响范围且具备恢复方案时执行。
不要在 YAML 中填写真实 API Key、AK 或 SK，应使用 Flink 加密变量或受控环境变量。
启用 sink.ignore-delete 或 sink.ignore-update-before 会保留源端已删除或旧主键的数据，使用前必须确认业务语义。
调整 Collection 名称模板后，同一源表可能写入新的 Collection；变更前应核对路由结果。

流式计算 Flink版

功能概述 #

版本兼容性 #

特色功能 #

前提条件 #

使用限制 #

语法结构 #

参数说明 #

连接与鉴权参数 #

Collection 与写入参数 #

连接参数 #

字段级参数 #

类型映射 #

使用示例 #

MySQL 数据同步到 VikingDB #

运行机制 #

验证与排障 #

验证方法 #

常见问题 #

注意事项 #

相关文档 #