Первоначальная синхронизация асинхронной репликации с помощью резервного копирования и восстановления

При этом способе первоначальной синхронизации исходные данные переносятся в приёмник из резервной копии в S3-совместимом хранилище. Затем асинхронная репликация применяет накопленные изменения и продолжает синхронизировать базы данных. Во время переноса приложения могут продолжать записывать данные в источник.

Этот способ позволяет использовать серверные операции резервного копирования и восстановления для переноса большого объёма данных. При первоначальном CDC-сканировании все исходные данные передаются через потоки изменений, а отдельная резервная копия не требуется. Различия в ресурсных и эксплуатационных затратах приведены в сравнении способов первоначальной синхронизации.

Подготовка

В примере таблица original_table базы /Root/source переносится в таблицу replica_table базы /Root/target. Экземпляр репликации my_replication создаётся в приёмнике. Команды используют профили YDB CLI source и target для подключения к соответствующим базам данных.

Перед началом работы:

  • Настройте профили и проверьте подключение к обеим базам с помощью команды scheme ls.
  • Подготовьте права для асинхронной репликации, экспорта в источнике и импорта в приёмнике. Пользователю, создающему экземпляр репликации, потребуется возможность изменять восстановленные таблицы для подключения их в качестве реплик.
  • В приёмнике создайте секрет source_token с токеном для подключения репликации к источнику.
  • Подготовьте бакет mybucket и аутентификацию в S3. В примерах используется эндпоинт storage.yandexcloud.net, а ключи доступа читаются из переменных окружения или файла ~/.aws/credentials. Эндпоинт должен быть доступен с серверов обеих баз данных: источник записывает резервную копию, приёмник читает её.
  • Выберите свободный префикс replication-bootstrap в бакете. Не изменяйте содержимое выгрузки до завершения импорта.
  • Убедитесь, что таблица replica_table отсутствует в приёмнике. Команда import s3 создаёт таблицу вместе с данными и не перезаписывает существующие таблицы. Проверить путь можно командой scheme ls.

На время первоначальной синхронизации сохраните состав реплицируемых таблиц, их столбцов и индексов. Схемы таблиц в резервной копии должны соответствовать схемам источника при запуске репликации.

Важно

До создания репликации выберите период доступности неподтверждённых изменений. При AVAILABILITY_PERIOD = 'UNLIMITED' изменения сохраняются до подтверждения обработки; обеспечьте место для их накопления в источнике. При конечном периоде планируйте экспорт, импорт и обработку накопленных изменений так, чтобы репликация успела подтвердить их обработку до истечения этого периода. Состояние PAUSED не останавливает отсчёт возраста сообщений.

1. Создайте репликацию без первоначального сканирования

Выполните в приёмнике запрос CREATE ASYNC REPLICATION с параметром INITIAL_SCAN = FALSE. YQL-запросы из этого руководства можно выполнить командой sql с профилем target.

CREATE ASYNC REPLICATION my_replication
FOR original_table AS replica_table
WITH (
    CONNECTION_STRING = 'grpcs://source.example.com:2135/?database=/Root/source',
    TOKEN_SECRET_PATH = 'source_token',
    INITIAL_SCAN = FALSE,
    AVAILABILITY_PERIOD = 'UNLIMITED'
);

Адрес source.example.com:2135 в примере соответствует эндпоинту источника. Параметры CONNECTION_STRING и TOKEN_SECRET_PATH задают подключение, которое приёмник будет использовать для создания и чтения потоков изменений.

В примере явно выбран период доступности 'UNLIMITED', который также используется по умолчанию. Для конечного периода укажите положительное значение типа Interval, например AVAILABILITY_PERIOD = Interval('PT48H'). Изменить этот параметр через ALTER ASYNC REPLICATION после создания нельзя.

Репликация создаётся в состоянии PAUSED и подготавливает потоки изменений в источнике без выгрузки первоначальных данных. Таблица replica_table пока не создаётся и не подключается к репликации. Её создаст операция импорта.

2. Дождитесь готовности потоков изменений

Создание потоков выполняется асинхронно. До запуска экспорта проверьте, что они готовы для всех реплицируемых таблиц, включая таблицы реализации индексов, если они входят в репликацию.

Получите описание репликации в формате JSON:

ydb -p target scheme describe my_replication --format proto-json-base64

Проверьте следующие поля:

  • paused: репликация ожидает запуска после импорта. Наличие этого поля само по себе не подтверждает готовность потоков.
  • items: перечень объектов с путями source_path и destination_path. Дождитесь появления всех ожидаемых объектов.
  • source_changefeed_name у каждого объекта: имя потока изменений. Имя назначается автоматически; не задавайте его вручную.

Для таблицы original_table проверьте состояние потока в источнике:

ydb -p source scheme describe original_table --format proto-json-base64

В массиве changefeeds найдите поток с именем из source_changefeed_name. Его поле state должно иметь значение STATE_ENABLED.

Для элемента, соответствующего синхронному или уникальному индексу, source_path указывает на индекс. Его таблица реализации находится по пути source_path/indexImplTable. Проверяйте массив changefeeds в описании этой таблицы.

Также проверьте параметры топика потока изменений: наличие читателя replicationConsumer, его политику доступности и значение RetentionPeriod. Путь к потоку зависит от объекта:

Объект Путь к потоку изменений
Таблица source_path/source_changefeed_name
Таблица реализации индекса source_path/indexImplTable/source_changefeed_name

В этих путях используйте значения полей из описания репликации. Для проверки выполните команду:

ydb -p source scheme describe <changefeed-path>

Где:

  • <changefeed-path>: путь к потоку изменений, например /Root/source/original_table/stream-name. Вместо stream-name используйте имя из описания репликации.

В таблице читателей для replicationConsumer проверьте:

  • При AVAILABILITY_PERIOD = 'UNLIMITED' поле Important должно иметь значение Yes.
  • При конечном периоде поле Important должно иметь значение No, а Availability period — соответствовать выбранному интервалу.

Повторите проверки для каждого объекта из items. Неподтверждённые сообщения защищены политикой читателя, поэтому оценивать доступность накопленных изменений только по RetentionPeriod нельзя. Подробнее см. Хранение изменений в источнике.

Важно

Запускайте экспорт только после готовности всех потоков изменений. Если резервная копия сделана раньше, изменения между её созданием и запуском потоков могут быть потеряны для репликации. Использовать произвольную ранее созданную резервную копию для этой процедуры нельзя.

3. Экспортируйте исходные данные

Запустите экспорт в S3 с профилем источника:

ydb -p source export s3 \
  --s3-endpoint storage.yandexcloud.net \
  --bucket mybucket \
  --destination-prefix replication-bootstrap \
  --include original_table \
  --include-index-data true

Выгрузка таблицы будет размещена под префиксом replication-bootstrap/original_table. Параметр --include-index-data true сохраняет также данные индексных таблиц, если у исходной таблицы есть индексы. На шаге импорта они будут загружены из резервной копии.

Для нескольких таблиц укажите --include для каждой из них в одной операции экспорта: сервер создаёт согласованную копию таблиц. Состав выгрузки должен соответствовать набору объектов репликации.

Команда возвращает идентификатор серверной операции. Дождитесь её успешного завершения с помощью operation get:

ydb -p source operation get <export-operation-id>

Где:

  • <export-operation-id>: идентификатор, полученный при запуске экспорта.

Переходите к импорту только при значениях ready = true, status = SUCCESS и progress = Done в табличном выводе. Успешный запуск команды экспорта ещё не означает завершение выгрузки.

Приложения могут продолжать изменять данные источника. Потоки изменений сохраняют эти изменения согласно выбранному AVAILABILITY_PERIOD, в том числе пока репликация находится в состоянии PAUSED.

4. Импортируйте данные в приёмник

После успешного экспорта запустите импорт из S3 с профилем приёмника:

ydb -p target import s3 \
  --s3-endpoint storage.yandexcloud.net \
  --bucket mybucket \
  --item src=replication-bootstrap/original_table,dst=replica_table \
  --index-population-mode import

Параметр --item сопоставляет префикс таблицы в резервной копии с целевым путём replica_table, указанным в CREATE ASYNC REPLICATION. Режим --index-population-mode import загружает данные индексов, сохранённые на предыдущем шаге. Для нескольких таблиц укажите отдельный --item для каждой пары путей.

Дождитесь успешного завершения операции с помощью operation get:

ydb -p target operation get <import-operation-id>

Где:

  • <import-operation-id>: идентификатор, полученный при запуске импорта.

Убедитесь, что ready = true, status = SUCCESS и progress = Done. Проверьте созданную таблицу:

ydb -p target scheme describe replica_table

До запуска репликации восстановленная таблица является обычной таблицей. Не записывайте в неё данные приложения: такие записи не синхронизируются с источником и могут быть перезаписаны при применении изменений.

5. Запустите репликацию изменений

После завершения импорта всех таблиц выполните в приёмнике запрос ALTER ASYNC REPLICATION:

ALTER ASYNC REPLICATION my_replication SET (STATE = 'STANDBY');

Приёмник проверит соответствие схем восстановленных таблиц источнику и подключит их как объекты-реплики. Проверка включает первичный ключ, столбцы, их типы, семейства столбцов и индексы. Подключённые таблицы становятся доступны только для чтения, а репликация начинает применять накопленные изменения.

Поддерживаются синхронные вторичные индексы, в том числе уникальные. Восстановленные векторные индексы удаляются при подключении таблицы к репликации. Подробнее см. Ограничения.

Записи читаются из потоков изменений, созданных перед экспортом. Среди них могут быть изменения, уже отражённые в резервной копии. Пока накопленные записи не обработаны, содержимое приёмника может изменяться и ещё не соответствует текущему состоянию источника.

Проверьте состояние и отставание:

ydb -p target scheme describe my_replication --stats

После запуска в описании поле paused сменяется полем running. Статистика содержит отставание репликации и отдельных объектов. Дождитесь обработки накопленных изменений; отсутствие значения отставания не означает, что оно равно нулю. Значение может быть недоступно при остановке читателя, потере его сессии или отсутствии свежих отчётов. Подробнее см. Репликация изменений.

После этого репликация продолжает переносить новые изменения. Если цель переноса состоит в миграции приложений, выполните отдельную процедуру завершения асинхронной репликации перед переключением пишущей нагрузки в приёмник.

Ошибки и повторные попытки

  • Если экспорт или импорт завершился ошибкой, проверьте статус и описание ошибки через operation get. Сохраняйте репликацию в состоянии PAUSED, пока импорт всех таблиц не завершится успешно. При повторном импорте учитывайте, что целевые таблицы не должны существовать; ошибка операции не является подтверждением отсутствия созданных объектов.
  • Если запуск репликации завершился ошибкой, получите JSON-описание репликации и проверьте поле error. В частности, проверьте наличие всех целевых таблиц, соответствие схем и права доступа. Не запускайте импорт поверх таблиц, которые уже подключены как реплики.
  • Если нужные CDC-записи удалены из-за истечения срока хранения, продолжение чтения не обеспечит полноту данных. Повторите первоначальную синхронизацию с новыми потоками изменений и новой резервной копией.
  • Перезапуск контроллера репликации сохраняет состояние PAUSED; он не заменяет явный перевод в STANDBY. При прерывании сессии CLI проверяйте ранее запущенные серверные операции по их идентификаторам, прежде чем запускать новые.

Подробнее о временных и критичных ошибках см. в разделе Обработка ошибок в процессе асинхронной репликации.

Предыдущая
Резервное копирование и восстановление
Следующая
Для разработчиков приложений