功能概述 #

Milvus Flink SQL Sink Connector 用于将火山引擎流式计算 Flink 版作业中的实时或批式计算结果写入向量数据库 Milvus 版。Connector 支持标量、向量、JSON 和 Array 字段，并提供 insert 与 upsert 两种写入语义。
本文仅适用于火山引擎流式计算 Flink 版。Connector 标识符为 milvus，已内置在对应的 Flink 发行版本中，无需额外上传 Maven 依赖或 Connector JAR。

版本兼容性 #

特色功能 #

• 支持将 FloatVector、BinaryVector、Float16Vector、BFloat16Vector 和 SparseFloatVector 等向量字段写入 Milvus。
• 支持通过单列主键执行 Upsert 和 Delete，适合接收 CDC 或其他 Changelog 数据。
• 支持写入 Milvus 动态字段。动态字段存储在隐藏的 JSON 字段 $meta 中，不会修改 Collection 的静态 Schema。
• 支持指定 partition-name，使 Insert、Upsert 和 Delete 在指定 Partition 内执行。

前提条件 #

1. 已创建 Milvus 实例、数据库和目标 Collection。
2. Collection 已定义主键、向量字段、标量字段及必要索引；目标 Partition 如需使用也已预先创建。
3. Flink 作业所在资源池能够访问 Milvus Endpoint。
4. 已获取实例私网地址、用户名和密码。按照向量数据库 Milvus 版的鉴权方式，将完整 Token 配置为 <用户名>:<密码>。
5. Flink DDL 的字段名称、大小写和类型与 Collection Schema 一致，并覆盖所有非 nullable 字段。
6. 使用 upsert 时，DDL 声明一个且仅一个主键，并与 Collection 主键字段一致。

建议在流式计算 Flink 版中创建加密变量保存完整 Token，并在 SQL 中通过 ${secret_values.milvus-token} 引用，避免明文保存用户名和密码。

使用限制 #

• Connector 不会自动创建 Collection、字段或索引，也不会执行 Schema 变更。
• insert 只接受 INSERT；收到 UPDATE 或 DELETE 时作业失败。
• upsert 接受 INSERT、UPDATE_AFTER 和 DELETE，并忽略 UPDATE_BEFORE；该模式要求单列主键。
• DELETE 仅支持整型或字符串主键，且主键不能为 NULL。
• Connector 不提供事务或两阶段提交，不支持端到端 Exactly-Once 外部写入。
• 配置 partition-name 时，目标 Partition 必须预先存在。该参数同时作用于 Insert、Upsert 和 Delete；Delete 只删除指定 Partition 中匹配主键的实体。
• 未配置 partition-name 时，Insert 写入 Milvus 的 _default Partition。使用 Upsert 时应保持分区配置稳定，避免同一主键分散在不同 Partition。
• Collection 未启用动态字段时，DDL 字段必须与 Collection 字段严格对应并区分大小写。
• Collection 启用动态字段后，DDL 中未在 Collection Schema 定义的列会写入 $meta，但不会变成静态 Schema 字段。
• 如需为现有 Collection 新增静态字段，Milvus 2.6 及以上版本可在 Connector 外部新增 nullable 标量字段，但不能以此方式新增向量字段；Milvus 2.5 及更早版本需要重建 Collection。
• Connector 启动时检查字段根类型；向量维度和 VarChar 长度等约束由 Milvus 服务端在写入时校验。

语法结构 #

CREATE TABLE milvus_sink (
    id BIGINT,
    embedding ARRAY<FLOAT>, 
    title STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'milvus',
    'uri' = '<milvus_uri>',
    'token' = '${secret_values.milvus-token}',
    'collection-name' = '<collection_name>',
    'sink.semantic' = 'upsert'
);

参数说明 #

通用参数 #

Sink 独有参数 #

类型映射 #

使用示例 #

追加写入 #

以下示例使用 insert 将仅包含 INSERT 的追加流写入 Milvus：

CREATE TABLE milvus_insert_sink (
    id BIGINT,
    embedding ARRAY<FLOAT>,
    title STRING
) WITH (
    'connector' = 'milvus',
    'uri' = '<milvus_uri>',
    'token' = '${secret_values.milvus-token}', 
    'collection-name' = '<collection_name>',
    'sink.semantic' = 'insert'
);

INSERT INTO milvus_insert_sink
SELECT id, embedding, title
FROM append_only_source;

预期结果：append_only_source 中的新增记录持续写入目标 Collection。上游一旦产生 UPDATE 或 DELETE，作业会失败。

同步更新和删除 #

以下示例使用 upsert 同步 Changelog 数据：

CREATE TABLE milvus_upsert_sink (
    id BIGINT,
    embedding ARRAY<FLOAT>,
    title STRING,
    PRIMARY KEY (id) NOT ENFORCED
) WITH (
    'connector' = 'milvus',
    'uri' = '<milvus_uri>',
    'token' = '${secret_values.milvus-token}',
    'collection-name' = '<collection_name>',
    'sink.semantic' = 'upsert',
    'sink.buffer-flush.max-rows' = '1000',
    'sink.buffer-flush.interval' = '2s'
);

INSERT INTO milvus_upsert_sink
SELECT id, embedding, title
FROM source_changelog;

预期结果： INSERT 和 UPDATE_AFTER 按主键写入或覆盖实体，DELETE 按主键删除实体，UPDATE_BEFORE 被忽略。

运行机制 #

Connector 在作业启动阶段读取 Collection Schema 并校验 DDL 字段。运行期间，Checkpoint、作业关闭、缓存行数阈值和定时刷新都会触发 Flush。
外部写入采用 At-Least-Once 语义。Connector 不对尚未提交的缓冲数据进行快照，故障恢复时由上游重放数据。对 CDC、聚合结果等存在重放可能的数据，建议配合稳定主键与 Upsert 写入，确保同一主键的最终结果幂等收敛。

验证与排障 #

验证方法 #

1. 提交 SQL 作业，确认作业进入 RUNNING，且启动日志中没有 Collection、字段或类型不兼容错误。
2. 写入一条新主键记录，在 Milvus 中按主键查询并核对向量维度和标量字段。
3. 使用 Upsert 时，对同一主键产生 UPDATE_AFTER 和 DELETE，确认实体先被覆盖、再被删除。
4. 等待至少一次 Checkpoint 成功后执行受控重启，确认作业恢复并继续写入；检查相同主键的最终值是否正确。

常见问题 #

注意事项 #

• enableDynamicField 不会为 Collection 自动增加静态字段；未声明字段只会保存到 $meta。
• 使用 partition-name 时，Insert、Upsert 和 Delete 必须保持一致的分区配置，且目标 Partition 必须预先存在。
• 生产作业不要明文配置 Milvus 用户名、密码或 Token。

相关文档 #

• 火山引擎：向量数据库 Milvus 版
• 火山引擎：连接 Milvus 实例
• 火山引擎：创建 Collection
• 火山引擎：加载和释放 Collection
• 火山引擎流式计算 Flink 版：使用变量
• 火山引擎流式计算 Flink 版：开发 Flink SQL 任务
• Milvus：启用动态字段
• Milvus：向现有 Collection 添加字段
• Milvus：插入、更新与删除实体