CDC 复制中的批量导入

本指南介绍了如何针对属于 CDC 复制拓扑的 Milvus 集群执行批量导入。在复制集群中，批量导入必须使用两阶段提交（2PC），以确保导入操作在主集群和备用集群上作为单个有序点被提交。

在本指南中，主集群指源 Milvus 集群，备集群指目标 Milvus 集群。

开始之前，请确保各集群之间已配置好 CDC 复制。有关详细信息，请参阅《设置 CDC 复制》。

为何需要 2PC

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

因此，请通过设置auto_commit=false ，以两阶段提交模式运行导入任务：

1. 导入阶段：Milvus 在主集群上加载数据并将导入任务复制到备用集群，但导入的数据仍不可见。导入任务停留在“Uncommitted ”状态并进入等待状态。

2. 提交阶段：您在主集群上显式提交导入任务。该提交将作为单个有序屏障复制到备用集群，因此两个集群都在同一逻辑点上使导入的数据可见。

步骤 1：在复制集群中启用导入

在复制集群中，导入功能默认处于禁用状态。请在主集群和备集群上将 `dataCoord.import.enableInReplicatingCluster ` 设置为 `true ` 以启用该功能。

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

spec:
  config:
    dataCoord:
      import:
        enableInReplicatingCluster: true

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

dataCoord:
  import:
    enableInReplicatingCluster: true

该设置支持热更新，因此无需完全重启即可生效。

启用此设置后，复制集群仅接受包含 `auto_commit=false` 的导入请求。下表列出了常见的被拒绝请求：

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

步骤 2：运行 2PC 导入

请在主集群上执行所有导入调用。导入的数据和提交决策会自动复制到备用集群，因此请勿在备用集群上自行提交或确认导入操作。

每个集群都会从各自的对象存储中读取导入文件。请确保待导入的文件同时存在于主集群和备用集群的对象存储中。您可以将文件上传到两个集群，或者使用两个集群均可读取的对象存储。如果备用集群上缺少文件，则该集群上的复制导入将因“对象未找到”错误而失败。

以下示例使用了来自pymilvus.bulk_writer 的基于 REST 的导入辅助函数。url 中的值即为您在其他 API 调用中使用的 Milvus 地址。

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")

为何要在两个集群上都等待Uncommitted

在备用集群完成导入之前进行提交不会导致数据损坏，但当提交被应用时，备用集群仍在进行数据同步。等待主集群和备用集群均报告Uncommitted ，可确保导入的数据已完全复制，且两个集群都已准备好共同显示该数据。

步骤 3：验证数据

当作业达到Completed 状态后，导入的实体在两个集群上均可见。在主集群上加载并查询该Collection，然后在备集群上运行相同的查询（无需在备集群上手动加载该Collection），并确认导入的实体在两个集群上均存在。

在处于备用状态期间，备用集群为只读模式。请勿直接在备用集群上提交导入、提交或其他 DDL 或 DCL 操作。应在主集群上执行这些操作，并让 CDC 复制将其应用到备用集群。

常见问题

应在哪个集群上执行导入和提交操作？

请在主集群上执行导入和提交操作。备用集群将通过 CDC 复制接收导入的数据和提交操作。

是否需要在备用集群上提交？

不需要。在主集群上提交后，系统会将该提交作为单个有序栅栏复制到备用集群。

为什么我的导入操作会因“import in replicating cluster is not supported yet ”而失败？

dataCoord.import.enableInReplicatingCluster 该集群上未启用此功能。请在主集群和备集群上均将其设置为“true ”。

为什么我的导入在启用auto_commit=true import in replicating cluster is not supported 时会失败？

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