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 」狀態後，匯入的實體將在兩個叢集上皆可見。請先在主叢集上載入並查詢該集合，接著在備用叢集上執行相同的查詢（無需在該處手動載入該集合），並確認匯入的實體確實存在於兩個叢集上。

備援叢集在維持備援狀態期間為唯讀模式。請勿直接在備援叢集上提交匯入、提交或其他 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"} 。