Massenimport bei der CDC-Replikation

In diesem Leitfaden wird erläutert, wie ein Massenimport für Milvus-Cluster durchgeführt wird, die Teil einer CDC-Replikationstopologie sind. In einem replizierenden Cluster muss der Massenimport das Two-Phase-Commit-Verfahren (2PC) verwenden, damit der Import als ein einziger, geordneter Punkt über den Primär- und den Standby-Cluster hinweg festgeschrieben wird.

In dieser Anleitung ist der Primärcluster der Milvus-Quellcluster und der Standby-Cluster der Milvus-Zielcluster.

Bevor Sie beginnen, stellen Sie sicher, dass die CDC-Replikation zwischen Ihren Clustern bereits konfiguriert ist. Weitere Informationen finden Sie unter „CDC-Replikation einrichten“.

Warum 2PC erforderlich ist

Ein normaler Massenimport wird automatisch bestätigt, sobald der Importauftrag abgeschlossen ist, wodurch die importierten Daten sofort sichtbar werden. In einer CDC-Replikationstopologie ist dieses Verhalten nicht zulässig, da der Primär- und der Standby-Cluster die importierten Daten zum selben logischen Zeitpunkt sichtbar machen müssen.

Führen Sie den Import stattdessen im Zwei-Phasen-Commit-Modus durch, indem Sie „ auto_commit=false “ festlegen:

1. Importphase: Milvus lädt die Daten auf dem Primärcluster und repliziert den Import auf den Standby-Cluster, doch die importierten Daten bleiben unsichtbar. Der Importjob hält im Status „ Uncommitted “ an und wartet.

2. Commit-Phase: Sie führen den Import-Job auf dem Primärcluster explizit ab. Der Commit wird als einzelner geordneter Fence auf den Standby-Cluster repliziert, sodass beide Cluster die importierten Daten am selben logischen Punkt sichtbar machen.

Schritt 1: Import in einem replizierenden Cluster aktivieren

Der Import in einem replizierenden Cluster ist standardmäßig deaktiviert. Aktivieren Sie ihn, indem Sie „ dataCoord.import.enableInReplicatingCluster “ sowohl auf dem Primär- als auch auf dem Standby-Cluster auf „ true “ setzen.

Wenn Sie Milvus mit dem Milvus Operator bereitstellen, fügen Sie die folgende Einstellung zu „ spec.config “ jeder „ Milvus “-Ressource hinzu:

spec:
  config:
    dataCoord:
      import:
        enableInReplicatingCluster: true

Wenn Sie Milvus direkt über „ milvus.yaml “ konfigurieren, fügen Sie die folgende Einstellung hinzu:

dataCoord:
  import:
    enableInReplicatingCluster: true

Diese Einstellung ist aktualisierbar und kann daher ohne vollständigen Neustart wirksam werden.

Wenn diese Einstellung aktiviert ist, akzeptiert ein replizierender Cluster nur Importe mit „ auto_commit=false “. Die folgende Tabelle listet häufige abgelehnte Anfragen auf:

Situation	Fehlermeldung
dataCoord.import.enableInReplicatingCluster ist nicht aktiviert	import in replicating cluster is not supported yet
auto_commit=true wurde übermittelt	auto_commit=true import in replicating cluster is not supported

Schritt 2: Führen Sie einen 2PC-Import durch

Führen Sie alle Importaufrufe auf dem Primärcluster aus. Die importierten Daten und die Commit-Entscheidung werden automatisch auf den Standby-Cluster repliziert. Übermitteln oder committen Sie den Import daher nicht selbst auf dem Standby-Cluster.

Jeder Cluster liest die Importdateien aus seinem eigenen Objektspeicher. Stellen Sie sicher, dass die zu importierenden Dateien sowohl im primären als auch im Standby-Objektspeicher vorhanden sind. Sie können die Dateien auf beide Cluster hochladen oder einen Objektspeicher verwenden, auf den beide Cluster zugreifen können. Fehlen die Dateien auf dem Standby-Cluster, schlägt der replizierte Import dort mit einem „Objekt nicht gefunden“-Fehler fehl.

Im folgenden Beispiel werden die REST-basierten Import-Helfer von pymilvus.bulk_writer verwendet. Die Werte für „ url “ entsprechen denselben Milvus-Adressen, die Sie auch für andere API-Aufrufe verwenden.

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

Warum auf beiden Clustern auf Uncommitted warten?

Ein Commit, bevor der Standby-Cluster den Import abgeschlossen hat, führt zwar nicht zu Datenkorruption, aber der Standby-Cluster ist zum Zeitpunkt der Commit-Anwendung noch dabei, den Rückstand aufzuholen. Das Warten, bis sowohl der Primär- als auch der Standby-Cluster „ Uncommitted “ melden, stellt sicher, dass die importierten Daten vollständig repliziert wurden und beide Cluster bereit sind, sie gemeinsam sichtbar zu machen.

Schritt 3: Überprüfen der Daten

Nachdem der Job den Status „ Completed “ erreicht hat, sind die importierten Entitäten auf beiden Clustern sichtbar. Laden Sie die Sammlung auf dem Primärcluster und führen Sie eine Abfrage darauf durch. Führen Sie anschließend dieselbe Abfrage auf dem Standby-Cluster aus, ohne die Sammlung dort manuell zu laden, und vergewissern Sie sich, dass die importierten Entitäten auf beiden Clustern vorhanden sind.

Der Standby-Cluster ist schreibgeschützt, solange er im Standby-Modus bleibt. Führen Sie keine Importe, Commits oder andere DDL- oder DCL-Operationen direkt auf dem Standby-Cluster durch. Führen Sie diese Operationen auf dem Primärcluster durch und lassen Sie sie durch die CDC-Replikation auf den Standby-Cluster anwenden.

FAQ

Auf welchem Cluster sollte ich den Import und das Commit ausführen?

Führen Sie den Import und das Commit auf dem Primärcluster durch. Der Standby-Cluster erhält sowohl die importierten Daten als auch das Commit über die CDC-Replikation.

Muss ich das Commit auf dem Standby-Cluster durchführen?

Nein. Durch das Committen auf dem Primärcluster wird das Commit als einzelner geordneter Fence auf den Standby-Cluster repliziert.

Warum schlägt mein Import mit der Meldung „ import in replicating cluster is not supported yet “ fehl?

dataCoord.import.enableInReplicatingCluster ist auf diesem Cluster nicht aktiviert. Setzen Sie die Einstellung sowohl auf dem Primär- als auch auf dem Standby-Cluster auf „ true “.

Warum schlägt mein Import mit „ auto_commit=true import in replicating cluster is not supported “ fehl?

In einem replizierenden Cluster werden nur 2PC-Importe mit „ auto_commit=false “ akzeptiert. Setzen Sie bei der Importanforderung „ options={"auto_commit": "false"} “.