Importação em massa na replicação CDC

Este guia explica como executar uma importação em massa em clusters Milvus que fazem parte de uma topologia de replicação CDC. Num cluster de replicação, a importação em massa deve utilizar o commit em duas fases (2PC), para que a importação seja confirmada como um único ponto ordenado nos clusters primário e de reserva.

Neste guia, o cluster primário é o cluster Milvus de origem e o cluster em espera é o cluster Milvus de destino.

Antes de começar, certifique-se de que a replicação CDC já está configurada entre os seus clusters. Para mais detalhes, consulte «Configurar a replicação CDC».

Por que razão o 2PC é necessário

Uma importação em massa normal é confirmada automaticamente quando a tarefa de importação termina, o que torna os dados importados visíveis imediatamente. Numa topologia de replicação CDC, este comportamento não é permitido porque os clusters primário e de reserva têm de tornar os dados importados visíveis no mesmo ponto lógico.

Em vez disso, execute a importação no modo de confirmação em duas fases, definindo « auto_commit=false »:

1. Fase de importação: o Milvus carrega os dados no cluster primário e replica a importação para o cluster de reserva, mas os dados importados permanecem invisíveis. A tarefa de importação pára no estado « Uncommitted » e fica em espera.

2. Fase de confirmação: confirma-se explicitamente a tarefa de importação no cluster primário. A confirmação é replicada para o cluster em espera como uma única barreira ordenada, pelo que ambos os clusters tornam os dados importados visíveis no mesmo ponto lógico.

Passo 1: Ativar a importação num cluster de replicação

A importação num cluster de replicação está desativada por predefinição. Ative-a definindo « dataCoord.import.enableInReplicatingCluster » como « true » tanto no cluster primário como no cluster em espera.

Se implementar o Milvus com o Milvus Operator, adicione a seguinte configuração a « spec.config » de cada recurso « Milvus »:

spec:
  config:
    dataCoord:
      import:
        enableInReplicatingCluster: true

Se configurar o Milvus diretamente através de ` milvus.yaml`, adicione a seguinte configuração:

dataCoord:
  import:
    enableInReplicatingCluster: true

Esta configuração é atualizável, pelo que pode entrar em vigor sem necessidade de um reinício completo.

Quando esta configuração está ativada, um cluster de replicação aceita apenas importações com ` auto_commit=false`. A tabela seguinte enumera os pedidos rejeitados mais comuns:

Situação	Mensagem de erro
dataCoord.import.enableInReplicatingCluster não está ativada	import in replicating cluster is not supported yet
auto_commit=true foi submetida	auto_commit=true import in replicating cluster is not supported

Passo 2: Executar uma importação 2PC

Execute todas as chamadas de importação no cluster primário. Os dados importados e a decisão de confirmação são replicados automaticamente para o cluster de reserva; por isso, não envie nem confirme a importação no cluster de reserva manualmente.

Cada cluster lê os ficheiros de importação a partir do seu próprio armazenamento de objetos. Certifique-se de que os ficheiros a importar existem tanto no armazenamento de objetos primário como no de reserva. Pode carregar os ficheiros para ambos os clusters ou utilizar um armazenamento de objetos que ambos os clusters possam ler. Se os ficheiros não estiverem presentes no cluster de reserva, a importação replicada falhará nesse cluster com um erro de objeto não encontrado.

O exemplo seguinte utiliza os auxiliares de importação baseados em REST disponíveis em pymilvus.bulk_writer. Os valores de url correspondem aos mesmos endereços Milvus que utiliza para outras chamadas de 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")

Porquê aguardar que Uncommitted esteja disponível em ambos os clusters

Efetuar o commit antes de o cluster de reserva ter concluído a importação não corrompe os dados, mas o cluster de reserva ainda está a atualizar-se quando o commit é aplicado. Esperar até que tanto o cluster primário como o de reserva indiquem « Uncommitted » confirma que os dados importados foram totalmente replicados e que ambos os clusters estão prontos para os tornar visíveis em simultâneo.

Passo 3: Verificar os dados

Depois de a tarefa atingir o estado « Completed », as entidades importadas ficam visíveis em ambos os clusters. Carregue e consulte a coleção no cluster primário; em seguida, execute a mesma consulta no cluster de reserva sem carregar manualmente a coleção nesse cluster e confirme que as entidades importadas estão presentes em ambos os clusters.

O cluster de reserva é de leitura apenas enquanto permanecer em modo de reserva. Não envie importações, confirmações nem outras operações DDL ou DCL diretamente para o cluster de reserva. Realize estas operações no cluster primário e deixe que a replicação CDC as aplique ao cluster de reserva.

Perguntas frequentes

Em que cluster devo executar a importação e a confirmação?

Execute a importação e a confirmação no cluster primário. O cluster em espera recebe tanto os dados importados como a confirmação através da replicação CDC.

Preciso de efetuar o commit no cluster em espera?

Não. O commit no cluster primário replica o commit para o cluster de reserva como uma única barreira ordenada.

Por que razão a minha importação falha com a mensagem « import in replicating cluster is not supported yet »?

dataCoord.import.enableInReplicatingCluster não está ativado nesse cluster. Defina-o para « true » tanto no cluster primário como no cluster de reserva.

Por que razão a minha importação falha com a opção « auto_commit=true import in replicating cluster is not supported »?

Num cluster de replicação, apenas são aceites importações 2PC com « auto_commit=false ». Defina « options={"auto_commit": "false"} » na solicitação de importação.