Milvus CDC 主备复制拓扑下的两阶段提交（2PC）批量导入实操指南-育师

Milvus CDC 主备复制拓扑下的两阶段提交（2PC）批量导入实操指南

【免费下载链接】milvusMilvus is a high-performance, cloud-native vector database built for scalable vector ANN search项目地址: https://gitcode.com/GitHub_Trending/mi/milvus

Milvus 自 v2.6 起可通过 CDC（Change Data Capture）构建主备（primary-standby）灾备拓扑，将主集群的变更实时复制到备集群。本文讲解在此类复制拓扑下如何安全地执行批量导入（bulk import）：复制集群必须使用两阶段提交（two-phase commit，2PC），导入数据先以不可见状态（Uncommitted）在主备两侧同时落地，再由你显式提交一次，使提交作为单一有序栅栏（fence）被复制到备集群，确保主备在同一逻辑点让数据可见。读完本文，你将掌握enableInReplicatingCluster配置的启用方法、2PC 导入的完整调用流程、状态轮询与验证手段，以及常见错误信息的排查思路。

为什么复制集群中的批量导入必须使用 2PC

普通批量导入是自动提交的：任务跑完后，数据自行变为可见。这在参与 CDC 复制的集群中是不允许的——主集群直接可见的时刻与备集群不一致，会破坏主备数据在同一逻辑点的对齐关系。

因此，复制集群中的导入必须以auto_commit=false进入两阶段提交模式：

导入阶段（Import phase）：数据在主集群被加载并通过 WAL 复制到备集群，但始终保持不可见。任务停在Uncommitted状态等待你显式触发提交。
提交阶段（Commit phase）：你在主集群显式提交该任务。提交以“单一有序栅栏”的形式被复制到备集群，两侧集群在同一个逻辑点同时让数据可见。

从源码看，这个两阶段状态机由 datacoord 的 import checker 驱动。任务完成数据与索引构建后先被切换到Uncommitted状态（import_checker.go 中注释为 "2PC: hand off to Uncommitted regardless of auto_commit"，并在该状态等待）。auto_commit=true的任务会由 checker 自动触发一次提交广播；而auto_commit=false的任务则等待平台通过显式的 CommitImport 调用触发提交（见 import_checker.go 中checkUncommittedJob的处理逻辑：注释明确写到"If auto_commit=false, it waits for an explicit CommitImport RPC from the platform"）。

前置条件

开始前请确认：

两个 Milvus 集群之间已正确配置 CDC 复制拓扑。若尚未搭建，请先参照 CDC Replication Quick Start 完成双集群部署并应用复制配置；整体角色行为可参考 CDC Replication Overview（备集群在作为备期间为只读角色，主集群负责读写并推送变更）。
主备两侧都已开启下文 Step 1 所述的导入开关。

Step 1：在复制集群中启用导入能力

复制集群中的导入默认是关闭的。需要将dataCoord.import.enableInReplicatingCluster配置设为true，并且主集群与备集群都要开启。

使用 Milvus Operator 部署时，在每个Milvus资源的spec.config中加入该设置：

spec: config: dataCoord: import: enableInReplicatingCluster: true

若直接通过milvus.yaml配置 Milvus，则写入：

dataCoord: import: enableInReplicatingCluster: true

仓库自带的默认配置 milvus.yaml 中该项默认值为false（enableInReplicatingCluster: false），注释与源码说明一致："Whether to allow import in a replicating cluster. When enabled, only auto_commit=false imports are accepted."

该配置是**可热更新（refreshable）**的：对应ParamItem在 component_param.go 中注册，开启后无需整体重启即可生效。从参数定义看，它的文档说明为："When enabled, only auto_commit=false imports are accepted"，即一旦开启，复制集群只接受auto_commit=false的导入。

不满足规则的导入会被拒绝

在复制集群中启用该开关后，凡违反上述规则的导入请求都会在提交阶段被直接拒绝并返回错误。仓库中的实际校验逻辑位于 datacoord 的 ddl_callbacks_import.go：validateImportReplication会先通过 WAL channel assignment 判断当前集群是否处于复制拓扑（isReplicatingCluster检查CrossClusterTopology非空或Clusters数量大于 1），随后依次校验开关与auto_commit取值。

触发场景	返回错误信息
处于复制拓扑但配置未开启	`import in replicating cluster is not supported yet`
配置已开启但提交了`auto_commit=true`	`auto_commit=true import in replicating cluster is not supported`

值得留意的是，该校验会发生两次：一次在请求校验阶段，另一次在真正向各 vchannel 广播导入消息前，持锁重新检查复制状态。这是为了防止在开关与锁窗口内 CDC 被临时开启，导致本应被拒绝的导入被广播进复制拓扑（参见 ddl_callbacks_import.go 中broadcastImport的注释）。

Step 2：运行一次 2PC 导入

所有导入调用都必须发往主集群（primary）。导入数据与提交决定会自动复制到备集群，因此不需要、也不允许在备集群上自行提交或发起导入。

每个集群都从自己的对象存储读取导入文件。因此必须保证待导入文件同时存在于主集群与备集群各自的对象存储中——要么分别上传到两侧，要么使用两个集群都能读取的共享对象存储。若文件在备集群侧缺失，复制过去的导入会在备集群侧以对象不存在（object-not-found）错误失败。

下面的示例使用pymilvus.bulk_writer提供的 REST 风格导入辅助函数，其中url即你平时调用 Milvus 各 API 时使用的地址：

import time from pymilvus.bulk_writer import ( bulk_import, get_import_progress, commit_import, ) # Primary and standby addresses. Replace with your own. source_url = "http://127.0.0.1:19530" target_url = "http://127.0.0.1:19531" collection_name = "demo_collection" # Object-storage paths of the files to import, prepared the same way as a # normal bulk import (for example, with BulkWriter). Each inner list is one # batch of files. files = [ ["import-data/part-1.parquet"], ] def wait_for_state(url, job_id, target_state, timeout=600): """Poll an import job until it reaches target_state (or fails).""" deadline = time.time() + timeout while time.time() < deadline: resp = get_import_progress(url=url, job_id=job_id) data = resp.json().get("data", {}) state = data.get("state") print(f"[{url}] job {job_id} state={state} progress={data.get('progress')}") if state == target_state: return if state == "Failed": raise RuntimeError( f"import job {job_id} failed on {url}: {data.get('reason')}" ) time.sleep(3) raise TimeoutError(f"job {job_id} did not reach {target_state} on {url}") # 1. Start a 2PC import on the PRIMARY. auto_commit=false is required in a # replicating cluster; the job stops at the Uncommitted state. resp = bulk_import( url=source_url, collection_name=collection_name, files=files, options={"auto_commit": "false"}, ) job_id = resp.json()["data"]["jobId"] print(f"started 2PC import job: {job_id}") # 2. Best practice: wait until BOTH clusters report Uncommitted before you # commit. The same job_id is used on the primary and the standby, because # the import is replicated through the WAL. wait_for_state(source_url, job_id, "Uncommitted") wait_for_state(target_url, job_id, "Uncommitted") # 3. Commit ONCE on the primary. The commit is replicated to the standby as a # single ordered fence, so you do not commit on the standby yourself. commit_import(url=source_url, job_id=job_id) print(f"committed import job: {job_id}") # 4. Wait for the job to complete on both clusters. wait_for_state(source_url, job_id, "Completed") wait_for_state(target_url, job_id, "Completed") print("import committed and visible on both clusters")

流程要点拆解如下：

options={"auto_commit": "false"}是硬性要求：复制集群不接受auto_commit=true的导入。
步骤 2 中主备两侧使用同一个job_id，因为导入本身是通过 WAL 复制过去的，job 在主备上天然同源。
服务端对提交动作的去重是安全的：即使提交消息在复制链路中出现重复送达，datacoord 的 commit 处理也会保证幂等（参见 ddl_callbacks_import.go 中commitImportV2AckCallback对 "job already committing or completed" 场景的 no-op 处理）。

为什么要等主备两侧都到达`Uncommitted`

在备集群尚未完成导入时就提前提交不会损坏数据，但意味着你提交的那一刻备集群仍在追赶中。等到主集群与备集群都报告Uncommitted，即可确认：导入数据已完成复制，两侧集群都准备好同步让数据可见。这样提交应用时，主备的差距保持在最小状态。

提交在服务端的落地路径

在主集群调用commit_import后，REST 请求会进入代理层的提交接口（见 handler_v2.go 中注册的CommitAction路由与其commitImportJob处理），由 datacoord 以 CommitImport WAL 广播的形式下发到各 vchannel。收到广播 ack 后，job 由Uncommitted推进到Committing状态，再在所有 vchannel 确认提交栅栏后转为Completed（见 import_checker.go 中checkCommittingJob的逻辑："Once all vchannels have acknowledged the commit fence, the job transitions to Completed"）。

Step 3：验证数据

当任务在主备两侧都到达Completed后，导入的数据行在两侧集群都可见。验证方式为：先在主集群加载（load）并查询该 collection，然后在不在备集群上手动执行 load的情况下对备集群执行相同查询，确认导入的行在两侧均存在。

注意：备集群在保持备角色期间是只读的。不要在备集群上直接提交导入、执行 commit 或其他 DDL / DCL 操作；这些操作应在主集群执行，交给复制机制去应用到备集群。

FAQ

我应该在哪个集群上发起导入与提交？

主集群。备集群通过复制同时收到导入数据与提交决定，你永远不要在备集群上提交或发起导入。

我需要在备集群上提交吗？

不需要。在主集群提交会把提交作为单一有序栅栏复制到备集群，备集群会在同一逻辑点让数据可见。

为什么我的导入报错 "import in replicating cluster is not supported yet"？

说明该集群尚未开启dataCoord.import.enableInReplicatingCluster。请在主集群与备集群上都设置为true（见上文 Step 1）。

为什么我的导入报错 "auto_commit=true import in replicating cluster is not supported"？

复制集群只接受auto_commit=false（即 2PC）的导入。请在导入请求中设置options={"auto_commit": "false"}。

从源码看：几个需要留意的实现细节

auto_commit开关位置：参数键为dataCoord.import.enableInReplicatingCluster，默认false，对应版本标注为 v2.7.0，配置项注册于 component_param.go。这是判定“复制拓扑是否允许导入”的唯一总开关。
复制拓扑的判定依据：datacoord 通过 stream channel 的最新 assignment 判断是否处于复制状态，其判定条件为复制配置中存在跨集群拓扑（CrossClusterTopology非空）或集群成员数大于 1（见 ddl_callbacks_import.go 中isReplicatingCluster）。也就是说，一旦解除复制配置，导入会自动回到普通自动提交语义。
主键一致性设计：复制集群中的 autoID 导入会由 datacoord 为每个文件预先分配主键范围（per-file PK ranges）并随 ImportMsg 下发，使主备两集群产生完全相同的主键值，而不是各自本地分配（见 ddl_callbacks_import.go 中assignPKRangesToFiles相关逻辑与注释）。
空导入的边界行为：对于没有任何数据的导入请求，auto-commit 路径会直接跳过Uncommitted转Completed；而复制集群的auto_commit=false路径则会进入Uncommitted状态等待平台提交（见 import_checker.go 中对应的分支注释）。
L0 导入与复制不兼容：仓库默认配置中dataCoord.import.enableL0Import默认关闭，原因正是恢复 L0 删除段与commit_timestamp（两阶段提交 / 复制导入）不兼容，可能静默破坏删除语义。2PC 导入场景下应避免 L0 导入。

Milvus CDC 主备复制拓扑下的两阶段提交（2PC）批量导入实操指南