Массовый импорт в репликации CDC

В данном руководстве объясняется, как выполнить массовый импорт в кластеры Milvus, входящие в топологию репликации CDC. В кластере с репликацией при массовом импорте необходимо использовать двухфазную фиксацию (2PC), чтобы импорт фиксировался как единая упорядоченная операция как в основном, так и в резервном кластерах.

В данном руководстве основной кластер — это исходный кластер Milvus, а резервный кластер — целевой кластер Milvus.

Прежде чем приступить к работе, убедитесь, что между вашими кластерами уже настроена репликация CDC. Подробности см. в разделе «Настройка репликации CDC».

Почему требуется 2PC

Обычный массовый импорт автоматически фиксируется по завершении задания импорта, благодаря чему импортированные данные становятся видимыми немедленно. В топологии репликации CDC такое поведение недопустимо, поскольку основной и резервный кластеры должны сделать импортированные данные видимыми в один и тот же логический момент времени.

Вместо этого запустите импорт в режиме двухфазной фиксации, установив параметр « auto_commit=false »:

1. Фаза импорта: Milvus загружает данные в основной кластер и реплицирует импорт в резервный кластер, но импортированные данные остаются невидимыми. Задание импорта останавливается в состоянии « Uncommitted » и переходит в режим ожидания.

2. Фаза фиксации: вы явно фиксируете задание импорта на первичном кластере. Фиксация реплицируется на резервный кластер в виде одного упорядоченного барьера, поэтому оба кластера делают импортированные данные видимыми в одной и той же логической точке.

Шаг 1. Включение импорта в кластере с репликацией

Импорт в кластере с репликацией по умолчанию отключен. Включите его, установив для параметра ` dataCoord.import.enableInReplicatingCluster ` значение ` true ` как в основном, так и в резервном кластере.

Если вы развертываете Milvus с помощью Milvus Operator, добавьте следующий параметр в файл ` spec.config ` каждого ресурса ` Milvus `:

spec:
  config:
    dataCoord:
      import:
        enableInReplicatingCluster: true

Если вы настраиваете Milvus напрямую через файл ` milvus.yaml`, добавьте следующий параметр:

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

Запустите все вызовы импорта на основном кластере. Импортированные данные и решение о фиксации автоматически реплицируются на резервный кластер, поэтому не отправляйте и не фиксируйте импорт на резервном кластере самостоятельно.

Каждый кластер считывает файлы импорта из собственного объектного хранилища. Убедитесь, что файлы для импорта присутствуют как в основном, так и в резервном объектном хранилище. Вы можете загрузить файлы в оба кластера или использовать объектное хранилище, доступное для чтения обоим кластерам. Если файлы отсутствуют в резервном кластере, репликация импорта в этом кластере завершится с ошибкой «объект не найден».

В следующем примере используются вспомогательные функции импорта на основе REST из pymilvus.bulk_writer. Значения url — это те же адреса Milvus, которые вы используете для других вызовов 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")

Почему нужно дождаться завершения 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 »?

В кластере с репликацией принимаются только импорты 2PC с параметром « auto_commit=false ». Укажите параметр « options={"auto_commit": "false"} » в запросе на импорт.