Importación masiva en la replicación CDC

Esta guía explica cómo realizar una importación masiva en clústeres de Milvus que forman parte de una topología de replicación CDC. En un clúster de replicación, la importación masiva debe utilizar la confirmación en dos fases (2PC) para que la importación se confirme como un único punto ordenado en los clústeres primario y de reserva.

En esta guía, el clúster primario es el clúster Milvus de origen, y el clúster de reserva es el clúster Milvus de destino.

Antes de empezar, asegúrate de que la replicación CDC ya esté configurada entre tus clústeres. Para obtener más información, consulta «Configurar la replicación CDC».

Por qué es necesario el 2PC

Una importación masiva normal se confirma automáticamente cuando finaliza el trabajo de importación, lo que hace que los datos importados sean visibles de inmediato. En una topología de replicación CDC, este comportamiento no está permitido porque los clústeres primario y de reserva deben hacer que los datos importados sean visibles en el mismo punto lógico.

En su lugar, ejecute la importación en modo de confirmación en dos fases configurando « auto_commit=false »:

1. Fase de importación: Milvus carga los datos en el clúster primario y replica la importación al clúster de reserva, pero los datos importados permanecen ocultos. El trabajo de importación se detiene en el estado « Uncommitted » y queda en espera.

2. Fase de confirmación: Se confirma explícitamente el trabajo de importación en el clúster primario. La confirmación se replica en el clúster de reserva como una única valla ordenada, de modo que ambos clústeres hacen visibles los datos importados en el mismo punto lógico.

Paso 1: Habilitar la importación en un clúster de replicación

La importación en un clúster de replicación está desactivada por defecto. Actívala configurando ` dataCoord.import.enableInReplicatingCluster ` como ` true ` tanto en el clúster primario como en el clúster de reserva.

Si implementa Milvus con Milvus Operator, añada la siguiente configuración a ` spec.config ` de cada recurso ` Milvus `:

spec:
  config:
    dataCoord:
      import:
        enableInReplicatingCluster: true

Si configura Milvus directamente a través de ` milvus.yaml`, añada la siguiente configuración:

dataCoord:
  import:
    enableInReplicatingCluster: true

Esta configuración se puede actualizar, por lo que puede surtir efecto sin necesidad de un reinicio completo.

Cuando esta configuración está habilitada, un clúster de replicación solo acepta importaciones con auto_commit=false. La siguiente tabla enumera las solicitudes rechazadas más habituales:

Situación	Mensaje de error
dataCoord.import.enableInReplicatingCluster no está habilitado	import in replicating cluster is not supported yet
auto_commit=true Se envía	auto_commit=true import in replicating cluster is not supported

Paso 2: Ejecutar una importación 2PC

Ejecuta todas las llamadas de importación en el clúster primario. Los datos importados y la decisión de confirmación se replican automáticamente en el clúster de reserva, por lo que no debes enviar ni confirmar la importación en el clúster de reserva por tu cuenta.

Cada clúster lee los archivos de importación desde su propio almacenamiento de objetos. Asegúrate de que los archivos que se van a importar existan tanto en el almacenamiento de objetos primario como en el de reserva. Puedes subir los archivos a ambos clústeres o utilizar un almacenamiento de objetos al que ambos clústeres puedan acceder. Si faltan archivos en el clúster de reserva, la importación replicada fallará allí con un error de «objeto no encontrado».

En el siguiente ejemplo se utilizan los ayudantes de importación basados en REST de pymilvus.bulk_writer. Los valores de url son las mismas direcciones de Milvus que utilizas para otras llamadas a la API.

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

¿Por qué esperar a que Uncommitted se ejecute en ambos clústeres?

Realizar la confirmación antes de que el clúster de reserva haya terminado de importar no daña los datos, pero el clúster de reserva aún se está poniendo al día cuando se aplica la confirmación. Esperar hasta que tanto el clúster primario como el de reserva informen de Uncommitted confirma que los datos importados se han replicado por completo y que ambos clústeres están listos para hacerlos visibles conjuntamente.

Paso 3: Verificar los datos

Una vez que el trabajo alcanza el estado « Completed », las entidades importadas son visibles en ambos clústeres. Carga y consulta la colección en el clúster primario; a continuación, ejecuta la misma consulta en el clúster de reserva sin cargar manualmente la colección allí y confirma que las entidades importadas están presentes en ambos clústeres.

El clúster de reserva es de solo lectura mientras permanece en estado de reserva. No envíes importaciones, confirmaciones ni otras operaciones DDL o DCL directamente al clúster de reserva. Realiza estas operaciones en el clúster primario y deja que la replicación CDC las aplique al clúster de reserva.

Preguntas frecuentes

¿En qué clúster debo ejecutar la importación y la confirmación?

Ejecute la importación y la confirmación en el clúster primario. El clúster de reserva recibe tanto los datos importados como la confirmación a través de la replicación CDC.

¿Tengo que realizar la confirmación en el clúster de reserva?

No. Al realizar la confirmación en el clúster primario, esta se replica en el clúster de espera como una única valla ordenada.

¿Por qué falla mi importación con el error « import in replicating cluster is not supported yet »?

dataCoord.import.enableInReplicatingCluster no está habilitado en ese clúster. Configúralo como « true » tanto en el clúster primario como en el de reserva.

¿Por qué falla mi importación con « auto_commit=true import in replicating cluster is not supported »?

En un clúster de replicación, solo se aceptan importaciones 2PC con « auto_commit=false ». Establece « options={"auto_commit": "false"} » en la solicitud de importación.