版本：v3.0.x

CDC Replication 中的 Bulk Import

本指南介绍如何对属于 CDC replication 拓扑的 Milvus 集群执行 Bulk Import。在复制集群中，Bulk Import 必须使用两阶段提交（2PC），以确保导入操作作为一个单一且有序的点在 primary 和 standby 集群之间提交。

在本指南中，primary cluster 指源 Milvus 集群，standby cluster 指目标 Milvus 集群。

开始之前，请确保已在集群之间配置 CDC replication。详情请参阅 Set Up CDC Replication。

Why 2PC is required

普通的 bulk import 会在导入任务完成时自动提交，使导入的数据立即可见。在 CDC 复制拓扑中，不允许这种行为，因为主集群和备用集群必须在同一个逻辑点使导入的数据可见。

因此，需要通过设置 auto_commit=false 以 two-phase commit 模式运行导入：

Import phase：Milvus 在主集群上加载数据，并将该导入复制到备用集群，但导入的数据仍不可见。导入任务会停在 Uncommitted 状态并等待。
Commit phase：你在主集群上显式提交导入任务。该提交会作为一个有序的单一 fence 复制到备用集群，因此两个集群会在同一个逻辑点使导入的数据可见。

Step 1: 在复制集群中启用 Import

复制集群中的 Import 默认处于禁用状态。你需要在 primary 和 standby 集群上都将 dataCoord.import.enableInReplicatingCluster 设置为 true 来启用该功能。

如果你使用 Milvus Operator 部署 Milvus，请将以下设置添加到每个 Milvus 资源的 spec.config 中：

YAML
spec:
  config:
    dataCoord:
      import:
        enableInReplicatingCluster: true

如果你直接通过 milvus.yaml 配置 Milvus，请添加以下设置：

YAML
dataCoord:
  import:
    enableInReplicatingCluster: true

该设置支持动态刷新，因此无需完整重启即可生效。

启用该设置后，复制集群只接受 auto_commit=false 的 Import。下表列出了常见的被拒绝请求：

情况	错误信息
未启用 `dataCoord.import.enableInReplicatingCluster`	`import in replicating cluster is not supported yet`
提交了 `auto_commit=true`	`auto_commit=true import in replicating cluster is not supported`

Step 2: Run a 2PC import

所有 import 调用都应在 primary cluster 上执行。导入的数据和 commit 决策会自动复制到 standby cluster，因此不要自行在 standby cluster 上提交或 commit 该 import。

每个集群都会从自己的对象存储中读取 import 文件。请确保要导入的文件同时存在于 primary 和 standby 的对象存储中。你可以将文件上传到两个集群，也可以使用两个集群都能读取的对象存储。如果 standby cluster 缺少这些文件，复制过去的 import 会在该集群上因 object-not-found 错误而失败。

以下示例使用 pymilvus.bulk_writer 中基于 REST 的 import helper。url 值与你用于其他 API 调用的 Milvus 地址相同。

Python
import time

from pymilvus.bulk_writer import bulk_import, commit_import, get_import_progress

primary_url = "http://127.0.0.1:19530"
standby_url = "http://127.0.0.1:19531"

collection_name = "demo_collection"

# Object-storage paths of the files to import. Prepare these files the same
# way as a normal bulk import, for example by using BulkWriter.
files = [
    ["import-data/part-1.parquet"],
]

def wait_for_state(url, job_id, target_state, timeout=600):
    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}")

# Start a 2PC import on the primary cluster. In a replicating cluster,
# auto_commit=false is required, and the job stops at the Uncommitted state.
resp = bulk_import(
    url=primary_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}")

# Wait until both clusters report Uncommitted. The same job ID is used on the
# primary and standby clusters because the import is replicated through CDC.
wait_for_state(primary_url, job_id, "Uncommitted")
wait_for_state(standby_url, job_id, "Uncommitted")

# Commit once on the primary cluster. Do not commit on the standby cluster.
commit_import(url=primary_url, job_id=job_id)
print(f"committed import job: {job_id}")

# Wait until the import is completed and visible on both clusters.
wait_for_state(primary_url, job_id, "Completed")
wait_for_state(standby_url, job_id, "Completed")
print("import committed and visible on both clusters")

Why wait for `Uncommitted` on both clusters

在 standby cluster 完成导入之前提交不会损坏数据，但提交生效时 standby cluster 仍在追赶进度。等到 primary 和 standby clusters 都报告 Uncommitted，可以确认导入的数据已完全复制，并且两个 clusters 都已准备好同时使其可见。

第 3 步：验证数据

当任务达到 Completed 状态后，导入的 Entity 会在两个集群上可见。先在主集群上 Load 并查询 Collection，然后在备用集群上运行相同查询，无需手动在备用集群上 Load 该 Collection，并确认两个集群上都存在已导入的 Entity。

备用集群在作为 standby 期间是只读的。不要直接向备用集群提交导入、commit 或其他 DDL 或 DCL 操作。请在主集群上执行这些操作，并让 CDC replication 将其应用到备用集群。

FAQ

应该在哪个 cluster 上运行 import 和 commit？

在 primary cluster 上运行 import 和 commit。standby cluster 会通过 CDC replication 接收导入的数据和 commit。

是否需要在 standby cluster 上 commit？

不需要。在 primary cluster 上 commit 后，该 commit 会作为一个有序的单一 fence 复制到 standby cluster。

为什么我的 import 会因 `import in replicating cluster is not supported yet` 而失败？

该集群未启用 dataCoord.import.enableInReplicatingCluster。请在 primary 和 standby 集群上都将其设置为 true。

为什么我的导入会因 `auto_commit=true import in replicating cluster is not supported` 而失败？

在正在复制的集群中，仅接受 auto_commit=false 的 2PC 导入。请在导入请求中设置 options={"auto_commit": "false"}。

Why 2PC is required​

Step 1: 在复制集群中启用 Import​

Step 2: Run a 2PC import​

Why wait for Uncommitted on both clusters​

第 3 步：验证数据​

FAQ​

应该在哪个 cluster 上运行 import 和 commit？​

是否需要在 standby cluster 上 commit？​

为什么我的 import 会因 import in replicating cluster is not supported yet 而失败？​

为什么我的导入会因 auto_commit=true import in replicating cluster is not supported 而失败？​