CDCレプリケーションにおける一括インポート

このガイドでは、CDCレプリケーショントポロジの一部であるMilvusクラスターに対して一括インポートを実行する方法について説明します。レプリケーションを行うクラスターでは、プライマリクラスターとスタンバイクラスター全体でインポートが単一の順序付きポイントとしてコミットされるよう、一括インポートでは2フェーズコミット（2PC）を使用する必要があります。

このガイドでは、プライマリ・クラスタをソース側の Milvus クラスタ、スタンバイ・クラスタをターゲット側の Milvus クラスタとします。

開始する前に、クラスタ間でCDCレプリケーションがすでに設定されていることを確認してください。詳細については、「CDCレプリケーションの設定」を参照してください。

2PCが必要な理由

通常のバルクインポートでは、インポートジョブが終了すると自動的にコミットされ、インポートされたデータが即座に表示されます。CDCレプリケーショントポロジーでは、プライマリクラスタとスタンバイクラスタがインポートされたデータを同じ論理的な時点で表示可能にする必要があるため、この動作は許可されません。

その代わりに、auto_commit=false を設定して、インポートを2フェーズコミットモードで実行してください：

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"} ` を設定してください。