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

Подробнее см. в документации Airbyte®.

Airbyte® уже встроен в Data Transfer, поэтому вам не нужно создавать отдельную виртуальную машину для его развертывания и добавления пользователя. Достаточно обеспечить сетевой доступ Data Transfer к базе-источнику.

Источник S3

Если вы используете частный бакет в качестве источника, предоставьте разрешения read и list учетной записи, которую будете использовать для подключения.

Подробнее см. в документации Airbyte®.

Источник Apache Kafka®

Managed Service for Apache Kafka®

Apache Kafka®

Создайте пользователя с ролью ACCESS_ROLE_CONSUMER на топике-источнике.

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Настройте доступ к кластеру-источнику из Yandex Cloud.
Настройте права доступа пользователя к нужному топику.

Выдайте права READ консьюмер-группе, идентификатор которой совпадает с идентификатором трансфера.

bin/kafka-acls --bootstrap-server localhost:9092 \
  --command-config adminclient-configs.conf \
  --add \
  --allow-principal User:username \
  --operation Read \
  --group <идентификатор_трансфера>

(Опционально) Чтобы использовать авторизацию по логину и паролю, настройте SASL-аутентификацию.

Источник ClickHouse®

Примечание

Yandex Data Transfer не может переносить базы данных ClickHouse®, в названии которых есть дефис.

При переносе таблиц с движками, отличными от движков на базе ReplicatedMergeTree и Distributed, в многохостовом кластере ClickHouse® трансфер завершится с ошибкой: the following tables have not Distributed or Replicated engines and are not yet supported.

Managed Service for ClickHouse®

ClickHouse®

Убедитесь, что переносимые таблицы используют движки семейства MergeTree. Будут перенесены только эти таблицы и материализованные представления (MaterializedView).

В случае многохостового кластера будут перенесены таблицы и материализованные представления только с движками на базе ReplicatedMergeTree либо Distributed. Проверьте, что данные таблицы и представления присутствуют на всех хостах кластера.
Создайте пользователя с доступом к базе источника. В настройках пользователя укажите для параметра Max execution time значение не менее 1000000.

Убедитесь, что переносимые таблицы используют движки семейства MergeTree. Будут перенесены только эти таблицы и материализованные представления (MaterializedView).

В случае многохостового кластера будут перенесены таблицы и материализованные представления только с движками на базе ReplicatedMergeTree либо Distributed. Проверьте, что данные таблицы и представления присутствуют на всех хостах кластера.
Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Настройте доступ к кластеру-источнику из Yandex Cloud.
Создайте пользователя с доступом к базе источника. В настройках пользователя укажите для параметра Max execution time значение не менее 1000000.

Источник Greenplum®

Примечание

Данные, хранящиеся в MATERIALIZED VIEW, не переносятся. Для переноса данных из MATERIALIZED VIEW создайте обыкновенный VIEW, ссылающийся на переносимый MATERIALIZED VIEW.

Yandex MPP Analytics for PostgreSQL

Greenplum®

Создайте пользователя, от имени которого трансфер подключится к источнику. Для этого выполните команду:
```
CREATE ROLE <имя_пользователя> LOGIN ENCRYPTED PASSWORD '<пароль>';
```
Настройте кластер-источник так, чтобы созданный пользователь мог подключаться ко всем хостам-мастерам кластера.
Если предполагается использовать параллельное копирование, настройте кластер-источник так, чтобы созданный пользователь мог подключаться ко всем хостам-сегментам кластера в режиме прямого доступа (utility mode). Для этого убедитесь, что для кластера включена настройка "Доступ из Data Transfer".
Выдайте созданному пользователю привилегию на выполнение операции SELECT над таблицами, которые переносит трансфер, и привилегию USAGE на схемы, в которых находятся эти таблицы.

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

Таблицы, для которых не выданы необходимые привилегии, недоступны для Data Transfer. Эти таблицы обрабатываются так же, как если бы они отсутствовали.

В этом примере привилегии выдаются на все таблицы в выбранной схеме:
```
GRANT SELECT ON ALL TABLES IN SCHEMA <название_схемы> TO <имя_пользователя>;
GRANT USAGE ON SCHEMA <название_схемы> TO <имя_пользователя>;
```

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Создайте пользователя, от имени которого трансфер подключится к источнику. Для этого выполните команду:
```
CREATE ROLE <имя_пользователя> LOGIN ENCRYPTED PASSWORD '<пароль>';
```
Настройте кластер-источник так, чтобы созданный пользователь мог подключаться ко всем хостам-мастерам кластера.
Если предполагается использовать параллельное копирование, настройте кластер-источник так, чтобы созданный пользователь мог подключаться ко всем хостам-сегментам кластера в режиме прямого доступа (utility mode).
Выдайте созданному пользователю привилегию на выполнение операции SELECT над таблицами, которые переносит трансфер, и привилегию USAGE на схемы, в которых находятся эти таблицы.

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

Таблицы, для которых не выданы необходимые привилегии, недоступны для Data Transfer. Эти таблицы обрабатываются так же, как если бы они отсутствовали.

В этом примере привилегии выдаются на все таблицы базы данных:
```
GRANT SELECT ON ALL TABLES IN SCHEMA <название_схемы> TO <имя_пользователя>;
GRANT USAGE ON SCHEMA <название_схемы> TO <имя_пользователя>;
```

Data Transfer взаимодействует с Greenplum® по-разному в зависимости от настроек трансфера и содержимого кластера-источника. Подробная информация доступна в разделе настройка эндпоинта-источника Greenplum®.

Источник MongoDB

Yandex StoreDoc (Managed Service for MongoDB)

MongoDB

Оцените общее количество баз данных для трансфера и общую нагрузку на Yandex StoreDoc. Если нагрузка на базы выше 10 000 операций на запись в секунду, создайте несколько эндпоинтов и трансферов. Подробнее см. в разделе Передача данных из эндпоинта-источника MongoDB/Yandex StoreDoc (Managed Service for MongoDB).
Создайте пользователя с ролью readWrite на каждую базу в источнике, которая будет реплицироваться. Роль readWrite нужна для того, чтобы у трансфера было право записи в служебную коллекцию __data_transfer.__dt_cluster_time.

Оцените общее количество баз данных для трансфера и общую нагрузку на MongoDB. Если нагрузка на базы выше 10 000 операций на запись в секунду, создайте несколько эндпоинтов и трансферов. Подробнее см. в разделе Передача данных из эндпоинта-источника MongoDB/Yandex StoreDoc (Managed Service for MongoDB).
Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Убедитесь, что версия MongoDB на приемнике не ниже 4.0.
Убедитесь, что кластер MongoDB сконфигурирован таким образом, чтобы на запросы к нему он возвращал корректно разрешающиеся IP-адреса или FQDN (Fully Qualified Domain Name).
Настройте доступ к кластеру-источнику из Yandex Cloud. Чтобы настроить кластер-источник для подключения из интернета:
1. Измените в конфигурационном файле значение настройки net.bindIp со 127.0.0.1 на 0.0.0.0:
```
# network interfaces
net:
  port: 27017
  bindIp: 0.0.0.0
```
2. Перезапустите сервис mongod:
```
sudo systemctl restart mongod.service
```
Если кластер-источник не использует репликацию, включите ее:
1. Добавьте в конфигурационный файл /etc/mongod.conf настройки репликации:
```
replication:
  replSetName: <имя_набора_реплик>
```
2. Перезапустите сервис mongod:
```
sudo systemctl restart mongod.service
```
3. Подключитесь к MongoDB и инициализируйте набор реплик командой:
```
rs.initiate({
    _id: "<имя_набора_реплик>",
    members: [{
        _id: 0,
        host: "<IP-адрес_который_слушает_Yandex_StoreDoc>:<порт>"
    }]
});
```
Создайте пользователя с ролью readWrite на все базы в источнике, которые будут реплицироваться:
```
use admin
db.createUser({
    user: "<имя_пользователя>",
    pwd: "<пароль>",
    mechanisms: ["SCRAM-SHA-1"],
    roles: [
        {
            db: "<имя_базы-источника_1>",
            role: "readWrite"
        },
        {
            db: "<имя_базы-источника_2>",
            role: "readWrite"
        },
        ...
    ]
});
```
После старта трансфер подключится к источнику от имени этого пользователя. Роль readWrite нужна для того, чтобы у трансфера было право записи в служебную коллекцию __data_transfer.__dt_cluster_time.

Примечание

Для версий MongoDB, начиная с 3.6, достаточно выдать созданному пользователю роль read на реплицируемые базы.
При использовании MongoDB, начиная с версии 3.6, для работы трансфера необходимо, чтобы пользователь обладал правами на чтение коллекции local.oplog.rs, а также на запись с чтением коллекции __data_transfer.__dt_cluster_time. Чтобы назначить пользователю роль clusterAdmin, предоставляющую такие права, подключитесь к MongoDB и выполните команды:
```
use admin;
db.grantRolesToUser("<имя_пользователя>", ["clusterAdmin"]);
```
Для выдачи более гранулярных прав можно назначить роль clusterMonitor, необходимую для чтения коллекции local.oplog.rs, а также дать доступ на чтение и запись системной коллекции __data_transfer.__dt_cluster_time.

Источник MySQL®

Managed Service for MySQL®

MySQL®

Включите режим полного бинарного лога на источнике, установив значение FULL или NOBLOB для параметра Binlog row image.
(Опционально) Настройте лимит на размер отправляемых кусков данных (chunk) с помощью параметра Max allowed packet.
Создайте пользователя для подключения к источнику.
1. Выдайте пользователю привилегию ALL_PRIVILEGES для базы-источника.
2. Выдайте пользователю административные привилегии REPLICATION CLIENT и REPLICATION SLAVE.
Для типов трансфера Репликация и Копирование и репликация таблицы без уникальных индексов не переносятся.

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

Чтобы сохранить работоспособность трансфера при переносе базы, содержащей таблицы без уникальных индексов:
- Не переносите такие таблицы. Для этого добавьте их в список исключенных таблиц в настройках эндпоинта для источника.
- Создайте уникальные индексы, например первичные ключи (PRIMARY KEY), в тех мигрируемых таблицах, где их нет.
  1. Чтобы получить список таблиц без первичного ключа, выполните запрос:
```
SELECT
    tab.table_schema AS database_name,
    tab.table_name AS table_name,
    tab.table_rows AS table_rows,
    tco.*
FROM information_schema.tables tab
    LEFT JOIN information_schema.table_constraints tco
        ON (tab.table_schema = tco.table_schema
            AND tab.table_name = tco.table_name
)
WHERE
    tab.table_schema NOT IN ('mysql', 'information_schema', 'performance_schema', 'sys')
    AND tco.constraint_type IS NULL
    AND tab.table_type = 'BASE TABLE';
```
  2. Изучите структуру таблиц без первичного ключа, которые необходимо перенести на приемник:
```
SHOW CREATE TABLE <имя_базы>.<имя_таблицы>;
```
  3. Добавьте простой или составной первичный ключ к таблицам, которые необходимо перенести на приемник:
```
ALTER TABLE <имя_таблицы> ADD PRIMARY KEY (<столбец_или_группа_столбцов>);
```
  4. Если в переносимой на приемник таблице нет столбца или группы столбцов, подходящих на роль первичного ключа, создайте новый столбец:
```
ALTER TABLE <имя_таблицы> ADD id BIGINT PRIMARY KEY AUTO_INCREMENT;
```
Примечание

Если создание первичного ключа завершается ошибкой Creating index 'PRIMARY' required more than 'innodb_online_alter_log_max_size' bytes of modification log. Please try again, увеличьте в настройках СУБД значение параметра Innodb log file size.
Выключите перенос триггеров на стадии активации трансфера и включите его на стадии деактивации (для типов трансфера Репликация и Копирование и репликация). Подробнее см. в описании дополнительных настроек эндпоинта для источника MySQL®.

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Убедитесь, что источник использует подсистему хранения данных низкого уровня MyISAM или InnoDB. При использовании других подсистем трансфер может завершиться с ошибкой.
Включите режим полного бинарного лога на источнике, установив значение FULL или NOBLOB для параметра binlog_row_image.
Задайте строковый формат бинарного лога на источнике, установив значение ROW для параметра binlog_format.
Для типов трансфера Репликация и Копирование и репликация:
- Укажите путь к расположению бинарного лог-файла в параметре log_bin.
- Выведите информацию о бинарном логе с помощью запроса SHOW MASTER STATUS (для версий MySQL® 5.7 и 8.0) или SHOW BINARY LOG STATUS (для версии MySQL® 8.4). Запрос должен возвращать строку с информацией, а не пустой ответ.
Если источник репликации — кластер, который находится за балансером, включите для него GTID-режим (GTID-MODE = ON).

Если по какой-то причине включить GTID-режим невозможно, убедитесь, что шаблон имен бинарных логов содержит имя хоста.

В обоих случаях это позволит продолжить репликацию в случае смены хоста-мастера.
(Опционально) Настройте лимит на размер отправляемых кусков данных (chunk) с помощью параметра max_allowed_packet.

Создайте пользователя для подключения к источнику и выдайте ему необходимые привилегии:

CREATE USER '<имя_пользователя>'@'%' IDENTIFIED BY '<пароль>';
GRANT ALL PRIVILEGES ON <имя_базы>.* TO '<имя_пользователя>'@'%';
GRANT REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO '<имя_пользователя>'@'%';

Для типов трансфера Репликация и Копирование и репликация таблицы без уникальных индексов не переносятся.

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

Чтобы сохранить работоспособность трансфера при переносе базы, содержащей таблицы без уникальных индексов:
- Не переносите такие таблицы. Для этого добавьте их в список исключенных таблиц в настройках эндпоинта для источника.
- Создайте уникальные индексы, например первичные ключи (PRIMARY KEY), в тех мигрируемых таблицах, где их нет.
  1. Чтобы получить список таблиц без первичного ключа, выполните запрос:
```
SELECT
    tab.table_schema AS database_name,
    tab.table_name AS table_name,
    tab.table_rows AS table_rows,
    tco.*
FROM information_schema.tables tab
    LEFT JOIN information_schema.table_constraints tco
        ON (tab.table_schema = tco.table_schema
            AND tab.table_name = tco.table_name
)
WHERE
    tab.table_schema NOT IN ('mysql', 'information_schema', 'performance_schema', 'sys')
    AND tco.constraint_type IS NULL
    AND tab.table_type = 'BASE TABLE';
```
  2. Изучите структуру таблиц без первичного ключа, которые необходимо перенести на приемник:
```
SHOW CREATE TABLE <имя_базы>.<имя_таблицы>;
```
  3. Добавьте простой или составной первичный ключ к таблицам, которые необходимо перенести на приемник:
```
ALTER TABLE <имя_таблицы> ADD PRIMARY KEY (<столбец_или_группа_столбцов>);
```
  4. Если в переносимой на приемник таблице нет столбца или группы столбцов, подходящих на роль первичного ключа, создайте новый столбец:
```
ALTER TABLE <имя_таблицы> ADD id BIGINT PRIMARY KEY AUTO_INCREMENT;
```
Примечание

Если создание первичного ключа завершается ошибкой Creating index 'PRIMARY' required more than 'innodb_online_alter_log_max_size' bytes of modification log. Please try again, увеличьте в настройках СУБД значение параметра inno_db_log_file_size.
Выключите перенос триггеров на стадии активации трансфера и включите его на стадии деактивации (для типов трансфера Репликация и Копирование и репликация). Подробнее см. в описании дополнительных настроек эндпоинта для источника MySQL®.

Источник Elasticsearch

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.

Источник OpenSearch

OpenSearch

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.

Источник Oracle

Примечание

В некоторых версиях Oracle для системных объектов вместо префикса V$ используются V_$. Например, V_$DATABASE вместо V$DATABASE.

Измените префиксы, если вы столкнулись с ошибкой вида can only select from fixed tables/views при выдаче прав на системные объекты.

Oracle

Чтобы подготовить источник к трансферу типа Копирование:
1. Создайте пользователя, от имени которого трансфер подключится к источнику:
```
CREATE USER <имя_пользователя> IDENTIFIED BY <пароль>;
GRANT CREATE SESSION TO <имя_пользователя>;
```
2. Выдайте права созданному пользователю:
```
GRANT SELECT ON V$DATABASE TO <имя_пользователя>;
GRANT SELECT ON DBA_EXTENTS TO <имя_пользователя>;
GRANT SELECT ON DBA_OBJECTS TO <имя_пользователя>;
GRANT FLASHBACK ANY TABLE TO <имя_пользователя>;
```
  При необходимости, право FLASHBACK можно выдать не на все таблицы (ANY TABLE), а только на те, которые нужно скопировать.
3. Выдайте пользователю права на чтение таблиц, которые нужно скопировать.

Чтобы подготовить источник к трансферу типа Репликация:

Создайте пользователя, от имени которого трансфер подключится к источнику:

CREATE USER <имя_пользователя> IDENTIFIED BY <пароль>;
ALTER USER <имя_пользователя> DEFAULT tablespace USERS TEMPORARY tablespace TEMP;
ALTER USER <имя_пользователя> quote unlimited on USERS;

GRANT 
    CREATE SESSION,
    execute_catalog_role,
    SELECT ANY TRANSACTION,
    SELECT ANY DISCTIONARY,
    CREATE PROCEDURE,
    LOGMINING 
TO <имя_пользователя>;

Выдайте права созданному пользователю:

GRANT SELECT ON V$DATABASE TO <имя_пользователя>;
GRANT SELECT ON V$LOG TO <имя_пользователя>;
GRANT SELECT ON V$LOGFILE TO <имя_пользователя>;
GRANT SELECT ON V$ARCHIVED_LOG TO <имя_пользователя>;

GRANT SELECT ON dba_objects TO <имя_пользователя>;
GRANT SELECT ON dba_extents TO <имя_пользователя>;

GRANT EXECUTE ON SYS.DBMS_LOGMNR TO <имя_пользователя>;
GRANT SELECT ON SYSTEM.LOGMNR_COL$ TO <имя_пользователя>;
GRANT SELECT ON SYSTEM.LOGMNR_OBJ$ TO <имя_пользователя>;
GRANT SELECT ON SYSTEM.LOGMNR_USER$ TO <имя_пользователя>;
GRANT SELECT ON SYSTEM.LOGMNR_UID$ TO <имя_пользователя>;

Выдайте пользователю права на чтение таблиц, которые нужно реплицировать.
Включите Minimal Supplemental Logging с первичными ключами, для этого выполните:
```
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;
```

Если вы работаете в CDB-среде, выполните следующие настройки:

Создайте пользователя Common User:

CREATE USER C##<имя_пользователя> IDENTIFIED BY <пароль> CONTAINER=all;
ALTER USER C##<имя_пользователя> DEFAULT TABLESPACE USERS temporary tablespace TEMP CONTAINER=all;
ALTER USER C##<имя_пользователя> quota unlimited on USERS CONTAINER=all;
ALTER USER C##<имя_пользователя> SET container_data = (cdb$root, <имя_вашей_PCB>) CONTAINER=current;

GRANT 
    CREATE SESSION,
    execute_catalog_role,
    SELECT ANY TRANSACTION,
    SELECT ANY DICTIONALY,
    CREATE PROCEDURE,
    LOGMINING,
    SET CONTAINER
TO C##<имя_пользователя> CONTAINER=ALL;

При необходимости, можно указать только контейнер cdb$root и контейнер с таблицами, которые нужно перенести.

Чтобы пользователь мог переключаться на контейнер cdb$root, выдайте ему права ALTER SESSION:
```
GRANT ALTER SESSION TO C##<имя_пользователя>;
```

Выдайте права созданному пользователю:

GRANT SELECT ON V$DATABASE TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON V$LOG TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON V$LOGFILE TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON V$ARCHIVED_LOG TO C##<имя_пользователя> CONTAINER=ALL;

GRANT SELECT ON dba_objects TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON dba_extents TO C##<имя_пользователя> CONTAINER=ALL;

GRANT EXECUTE ON SYS.DBMS_LOGMNR TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON SYSTEM.LOGMNR_COL$ TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON SYSTEM.LOGMNR_OBJ$ TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON SYSTEM.LOGMNR_USER$ TO C##<имя_пользователя> CONTAINER=ALL;
GRANT SELECT ON SYSTEM.LOGMNR_UID$ TO C##<имя_пользователя> CONTAINER=ALL;

Источник PostgreSQL

Примечание

При трансфере из PostgreSQL в любой тип приемника объекты типа large objects не переносятся.

При переносе данных с типом TIMESTAMP WITHOUT TIME ZONE применяется часовой пояс, указанный в параметре timezone базы данных источника PostgreSQL.

Пример

В колонке с типом данных TIMESTAMP WITHOUT TIME ZONE записано значение 1970-01-01 00:00:00. То, как при трансфере будет перенесено значение, зависит от параметра timezone в БД:

Если значение параметра равно Etc/UTC, то время переносится как 1970-01-01 00:00:00+00.
Если значение параметра равно Europe/Moscow, то время переносится как 1970-01-01 00:00:00+03.

Если определение переносимого VIEW содержит вызов VOLATILE функции, то трансфер читает данные из такого VIEW с уровнем изоляции READ UNCOMMITTED. Консистентность между данными в этом VIEW и данными других переносимых объектов не гарантируется. Чтение из MATERIALIZED VIEW в определении VIEW равносильно вызову VOLATILE функции.

Большие объекты в системе хранения TOAST и данные с типом bytea переносятся без ограничений.

Managed Service for PostgreSQL

PostgreSQL

Настройте пользователя, от имени которого трансфер подключится к источнику:
1. Создайте пользователя.
2. Для типов трансфера Репликация и Копирование и репликация назначьте роль mdb_replication этому пользователю.
3. Подключитесь к базе данных, которую нужно мигрировать, от имени владельца базы и настройте привилегии:
  - SELECT над всеми таблицами базы данных, которые переносит трансфер.
  - SELECT над всеми последовательностями базы данных, которые переносит трансфер.
  - USAGE на схемы этих таблиц и последовательностей.
  - ALL PRIVILEGES (CREATE и USAGE) на задаваемую параметром эндпоинта схему служебных таблиц __consumer_keeper и __data_transfer_mole_finder, если эндпоинт будет использоваться для типов трансфера Репликация или Копирование и репликация.
Настройте количество подключений пользователя к базе данных.
Если источник репликации — кластер, включите для него расширение pg_tm_aux. Это позволит продолжить репликацию в случае смены хоста-мастера. В некоторых случаях при смене мастера в кластере трансфер может завершиться ошибкой. Подробнее см. в разделе Решение проблем.
Для типов трансфера Репликация и Копирование и репликация таблицы без первичных ключей не переносятся. Чтобы сохранить работоспособность трансфера при переносе базы с такими таблицами:
- Не переносите таблицы без первичных ключей. Для этого добавьте их в список исключенных таблиц в настройках эндпоинта для источника.
- Добавьте идентификатор реплики на таблицах без primary key:
  - Для таблиц с индексом установите REPLICA IDENTITY по unique key:
```
ALTER TABLE MY_TBL REPLICA IDENTITY USING INDEX MY_IDX;
```
  - Для таблиц без индекса измените REPLICA IDENTITY:
```
ALTER TABLE MY_TBL REPLICA IDENTITY FULL;
```
    В этом случае Data Transfer будет воспринимать такие таблицы как таблицы, в которых первичный ключ является составным, и в него входят все колонки таблицы.
Если в таблице нет первичных ключей, тогда в логической репликации не будет событий изменений строк (UPDATE, DELETE).
- Во время трансфера из PostgreSQL в PostgreSQL, если у вас в источнике трансфера не будет исключена таблица без первичных ключей, то вы увидите ошибку:
```
 failed to run (abstract1 source): Cannot parse logical replication message: failed to reload schema: primary key check failed: Tables: n / N check failed: "public"."MY_TBL": no key columns found
```
- Во время трансфера из PostgreSQL в другую базу данных, если у вас будет добавлена таблица без первичных ключей, то вы увидите ошибку:
```
failed to run (abstract1 source): Cannot parse logical replication message: failed to reload schema: primary key check failed: Tables: n / N check failed:
"public"."MY_TBL": no key columns found
```
Выключите перенос внешних ключей на стадии создания эндпоинта-источника. Создайте их заново после окончания трансфера.
Если база данных содержит таблицы, в которых есть генерируемые столбцы, то такие таблицы не переносятся и трансфер завершается ошибкой. Подробнее см. в разделе Решение проблем. Чтобы сохранить работоспособность трансфера при переносе базы данных с такими таблицами, добавьте их в список исключенных таблиц в настройках эндпоинта для источника.
Найдите и завершите слишком долгие DDL-запросы. Для этого сделайте выборку из системной таблицы PostgreSQL pg_stat_activity:
```
SELECT NOW() - query_start AS duration, query, state
FROM pg_stat_activity
WHERE state != 'idle' ORDER BY 1 DESC;
```
Будет возвращен список запросов, выполняющихся на сервере. Обратите внимание на запросы с высоким значением duration.
Выключите перенос триггеров на стадии активации трансфера и включите его на стадии деактивации (для типов трансфера Репликация и Копирование и репликация). Подробнее см. в описании дополнительных настроек эндпоинта для источника PostgreSQL.
Для параллельного чтения из таблицы по диапазонам убедитесь, что указан первичный ключ. После этого укажите количество воркеров и потоков в блоке Среда выполнения в параметрах трансфера.
Настройте мониторинг WAL-лога.

Для трансферов типа Репликация и Копирование и репликация используется логическая репликация. Для этого сам трансфер создает слот репликации со значением slot_name, равным идентификатору трансфера, который можно получить, выбрав трансфер в списке ваших трансферов. WAL может расти по разным причинам: из-за долгой транзакции или из-за проблемы на трансфере. Поэтому рекомендуется настроить мониторинг WAL-лога на стороне источника.
1. Для мониторинга размера использованного хранилища или диска настройте алерт средствами мониторинга (см. описание disk.used_bytes).
2. Задайте максимальный размер WAL при репликации в настройке Max slot wal keep size. Редактирование данной настройки доступно начиная с 13 версии PostgreSQL. Если вы хотите экстренно запретить операции чтения трансферу, то удалите слот репликации.
  
  Важно
  
  При значении настройки -1 (размер не ограничен) открытые логические слоты репликации, из которых не считывается информация, будут препятствовать удалению WAL-файлов. В результате WAL-файлы займут все дисковое пространство и вы потеряете возможность подключаться к кластеру.
3. Настройте алерт средствами Yandex Monitoring на метрику, которая используется для Total size of WAL files. Пороговые значения должны быть меньше, чем указаны для метрики disk.used_bytes, так как на диске, кроме данных, хранятся временные файлы, WAL-лог и другие типы данных. Текущий размер слота можно мониторить средствами БД через запрос, указав правильный slot_name, равный идентификатору трансфера:
```
SELECT slot_name, pg_size_pretty(pg_current_wal_lsn() - restart_lsn), active_pid, catalog_xmin, restart_lsn, confirmed_flush_lsn
FROM pg_replication_slots WHERE slot_name = '<идентификатор_трансфера>'
```

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Настройте пользователя, от имени которого трансфер подключится к источнику:
1. Создайте нового пользователя:
  - Для типа трансфера Копирование создайте пользователя командой:
```
CREATE ROLE <имя_пользователя> LOGIN ENCRYPTED PASSWORD '<пароль>';
```
  - Для типов трансфера Репликация и Копирование и репликация создайте пользователя с привилегией REPLICATION командой:
```
CREATE ROLE <имя_пользователя> WITH REPLICATION LOGIN ENCRYPTED PASSWORD '<пароль>';
```
2. Выдайте созданному пользователю привилегию на выполнение операции SELECT над всеми таблицами базы данных, которые переносит трансфер:
```
GRANT SELECT ON ALL TABLES IN SCHEMA <название_схемы> TO <имя_пользователя>;
```
3. Выдайте созданному пользователю привилегию на схемы переносимой базы данных:
  - Для типа трансфера Копирование выдайте привилегию USAGE:
```
GRANT USAGE ON SCHEMA <название_схемы> TO <имя_пользователя>;
```
  - Для типа трансфера Репликация и Копирование и репликация выдайте привилегии CREATE и USAGE (ALL PRIVILEGES), необходимые для создания служебных таблиц:
```
GRANT ALL PRIVILEGES ON SCHEMA <название_схемы> TO <имя_пользователя>;
```
4. Выдайте созданному пользователю привилегию SELECT на все последовательности базы данных, которые переносит трансфер:
```
GRANT SELECT ON ALL SEQUENCES IN SCHEMA <название_схемы> TO <имя_пользователя>;
```
5. Выдайте созданному пользователю привилегию CONNECT, если настройки кластера-источника по умолчанию не позволяют выполнять подключение для новых пользователей:
```
GRANT CONNECT ON DATABASE <название_базы_данных> TO <имя_пользователя>;
```
Настройте конфигурацию PostgreSQL:
1. Внесите изменения в конфигурацию и настройки аутентификации кластера-источника. Для этого отредактируйте файлы postgresql.conf и pg_hba.conf (на дистрибутивах Debian и Ubuntu они по умолчанию расположены в каталоге /etc/postgresql/<версия_PostgreSQL>/main/):
  1. Задайте максимальное количество подключений пользователя. Для этого в файле postgresql.conf отредактируйте параметр max_connections:
```
max_connections = <количество_подключений>
```
    Где <количество_подключений> — максимальное число подключений. Подробнее о том, как настроить этот параметр см. в Особенности работы с эндпоинтами.
    
    Текущее количество подключений вы можете посмотреть в системной таблице pg_stat_activity:
```
SELECT count(*) FROM pg_stat_activity;
```
  2. Установите уровень логирования для Write Ahead Log (WAL). Для этого установите значение logical для настройки wal_level в postgresql.conf:
```
wal_level = logical
```
  3. (Опционально) Настройте SSL: это поможет не только шифровать данные, но и сжимать их. Чтобы включить использование SSL, задайте нужное значение в файле postgresql.conf:
```
ssl = on
```
  4. Разрешите подключение к кластеру. Для этого отредактируйте параметр listen_addresses в файле postgresql.conf. Например, чтобы кластер-источник принимал запросы на подключение со всех IP-адресов:
```
listen_addresses = '*'
```
  5. Настройте аутентификацию в файле pg_hba.conf:
    SSL
    
    Без SSL
    hostssl all all <IP-адрес_подключения> md5 hostssl replication all <IP-адрес_подключения> md5
    host all all <IP-адрес_подключения> md5 host replication all <IP-адрес_подключения> md5
    Где <IP-адрес_подключения> может быть как точным IP-адресом, так и диапазоном IP-адресов. Например, чтобы разрешить доступ из сети Yandex Cloud, вы можете указать все публичные IP-адреса Yandex Cloud.
2. Если в кластере-источнике работает файрвол, разрешите входящие соединения с нужных адресов.
3. Чтобы применить настройки, перезапустите сервис PostgreSQL:
```
sudo systemctl restart postgresql
```
4. Проверьте статус сервиса PostgreSQL после перезапуска:
```
sudo systemctl status postgresql
```
Установите и включите расширение wal2json.
- Linux
  1. Подключите официальный репозиторий PostgreSQL для вашего дистрибутива.
  2. Обновите список доступных пакетов и установите пакет wal2json для используемой версии PostgreSQL.
- Windows 10, 11
  1. Если у вас не установлена Microsoft Visual Studio, загрузите и установите ее. Для сборки расширения wal2json достаточно редакции Community Edition. При установке выберите компоненты:
    - MSBuild,
    - MSVC v141 x86/x64 build tools,
    - C++\CLI support for v141 build tools,
    - MSVC v141 — VS 2017 C++ x64\x86 build tools,
    - MSVC v141 — VS 2017 C++ x64\x86 Spectre-mitigated libs,
    - самая свежая версия Windows SDK для используемой версии ОС,
    - прочие зависимости, которые устанавливаются автоматически для выбранных компонентов.
    Запомните номер устанавливаемой версии Windows SDK — он понадобится при указании параметров сборки wal2json.
  2. Загрузите исходный код wal2json со страницы проекта.
  3. Распакуйте архив с исходным кодом в каталог C:\wal2json\.
  4. Перейдите в каталог C:\wal2json.
  5. В рамках одной сессии PowerShell внесите изменения в файл wal2json.vcxproj:
    - замените строки C:\postgres\pg103 на путь к каталогу с установленной версией PostgreSQL, например:
      (Get-Content .\wal2json.vcxproj).replace('C:\postgres\pg103', 'C:\PostgreSQL\14') | ` Set-Content .\wal2json.vcxproj
    - замените параметр сборки /MP на /MT, например:
      (Get-Content .\wal2json.vcxproj).replace('/MP', '/MT') | Set-Content .\wal2json.vcxproj
    - укажите в параметре <WindowsTargetPlatformVersion> номер версии установленного компонента Windows SDK:
      (Get-Content .\wal2json.vcxproj).replace('<WindowsTargetPlatformVersion>8.1', '<WindowsTargetPlatformVersion><установленная_версия_Windows_SDK>') | ` Set-Content .\wal2json.vcxproj
    1. Укажите значение переменной окружения, необходимой для сборки wal2json, например, для Visual Studio Community Edition 2022:
      $VCTargetsPath='C:\Program Files\Microsoft Visual Studio\2022\Comminuty\MSBuild\Microsoft\VC\v150'
    2. Запустите сборку:
      & 'C:\Program Files\Microsoft Visual Studio\2022\Community\MSBuild\Current\Bin\MSBuild.exe' /p:Configuration=Release /p:Platform=x64
    3. Скопируйте файл wal2json.dll из каталога build/release в каталог lib установленной версии PostgreSQL.
Если источник репликации — кластер, установите и включите на его хостах расширение pg_tm_aux. Это позволит продолжить репликацию в случае смены хоста-мастера. В некоторых случаях при смене мастера в кластере трансфер может завершиться ошибкой. Подробнее см. в разделе Решение проблем.
Для типов трансфера Репликация и Копирование и репликация таблицы без первичных ключей не переносятся. Чтобы сохранить работоспособность трансфера при переносе базы с такими таблицами:
- Не переносите таблицы без первичных ключей. Для этого добавьте их в список исключенных таблиц в настройках эндпоинта для источника.
- Добавьте идентификатор реплики на таблицах без primary key:
  - Для таблиц с индексом установите REPLICA IDENTITY по unique key:
```
ALTER TABLE MY_TBL REPLICA IDENTITY USING INDEX MY_IDX;
```
  - Для таблиц без индекса измените REPLICA IDENTITY:
```
ALTER TABLE MY_TBL REPLICA IDENTITY FULL;
```
    В этом случае Data Transfer будет воспринимать такие таблицы как таблицы, в которых первичный ключ является составным, и в него входят все колонки таблицы.
Если в таблице нет первичных ключей, тогда в логической репликации не будет событий изменений строк (UPDATE, DELETE).
- Во время трансфера из PostgreSQL в PostgreSQL, если у вас в источнике трансфера не будет исключена таблица без первичных ключей, то вы увидите ошибку:
```
 failed to run (abstract1 source): Cannot parse logical replication message: failed to reload schema: primary key check failed: Tables: n / N check failed: "public"."MY_TBL": no key columns found
```
- Во время трансфера из PostgreSQL в другую базу данных, если у вас будет добавлена таблица без первичных ключей, то вы увидите ошибку:
```
failed to run (abstract1 source): Cannot parse logical replication message: failed to reload schema: primary key check failed: Tables: n / N check failed:
"public"."MY_TBL": no key columns found
```
Выключите перенос внешних ключей на стадии создания эндпоинта-источника. Создайте их заново после окончания трансфера.
Если база данных содержит таблицы, в которых есть генерируемые столбцы, то такие таблицы не переносятся и трансфер завершается ошибкой. Подробнее см. в разделе Решение проблем. Чтобы сохранить работоспособность трансфера при переносе базы данных с такими таблицами, добавьте их в список исключенных таблиц в настройках эндпоинта для источника.
Найдите и завершите слишком долгие DDL-запросы. Для этого сделайте выборку из системной таблицы PostgreSQL pg_stat_activity:
```
SELECT NOW() - query_start AS duration, query, state
FROM pg_stat_activity
WHERE state != 'idle' ORDER BY 1 DESC;
```
Будет возвращен список запросов, выполняющихся на сервере. Обратите внимание на запросы с высоким значением duration.
Выключите перенос триггеров на стадии активации трансфера и включите его на стадии деактивации (для типов трансфера Репликация и Копирование и репликация). Подробнее см. в описании дополнительных настроек эндпоинта для источника PostgreSQL.
Для параллельного чтения из таблицы по диапазонам убедитесь, что указан первичный ключ. После этого укажите количество воркеров и потоков в блоке Среда выполнения в параметрах трансфера.
Если на источнике настроена репликация через Patroni, добавьте в его конфигурацию блок ignore_slots:
```
ignore_slots:
  - database: <база_данных>
    name: <слот_репликации>
    plugin: wal2json
    type: logical
```
Где:
- database — имя базы данных, для которой настроен трансфер.
- name — имя слота репликации.
Имя базы данных и имя слота репликации должны совпадать со значениями, указанными в настройках эндпоинта для источника. По умолчанию имя слота репликации совпадает с ID трансфера.

В противном случае начало этапа репликации завершится ошибкой:
```
Warn(Termination): unable to create new pg source: Replication slotID <имя_слота_репликации> does not exist.
```
Настройте мониторинг WAL-лога. Для трансферов типа Репликация и Копирование и репликация используется логическая репликация. Для этого сам трансфер создает слот репликации со значением slot_name, равным идентификатору трансфера, который можно получить, выбрав трансфер в списке ваших трансферов. WAL может расти по разным причинам: из-за долгой транзакции или из-за проблемы на трансфере. Поэтому рекомендуется настроить мониторинг WAL-лога на стороне источника.
1. Настройте алерты на основе рекомендаций по использованию диска.
2. Установите максимальный размер WAL. Эта возможность доступна, начиная с версии PostgreSQL 13.
3. Текущий размер слота можно отслеживать средствами БД через запрос, указав правильный slot_name, равный идентификатору трансфера:
```
SELECT slot_name, pg_size_pretty(pg_current_wal_lsn() - restart_lsn), active_pid, catalog_xmin, restart_lsn, confirmed_flush_lsn
FROM pg_replication_slots WHERE slot_name = '<идентификатор_трансфера>'
```

Примечание

Об особенностях переноса данных из PostgreSQL в ClickHouse® трансферами типа Репликация и Копирование и репликация см. в разделе Асинхронная репликация данных из PostgreSQL в ClickHouse®.

Источник Yandex Data Streams

Создайте сервисный аккаунт с ролью yds.editor.
Создайте поток данных.

(Опционально) Создайте функцию обработки.

Пример функции обработки

const yc = require("yandex-cloud");
const { Parser } = require("@robojones/nginx-log-parser");
module.exports.handler = async function (event, context) {
    const schema =
        '$remote_addr - $remote_user [$time_local] "$request" $status $bytes_sent "$http_referer" "$http_user_agent"';
    const parser = new Parser(schema);
    return {
        Records: event.Records.map((record) => {
            const decodedData = new Buffer(record.kinesis.data, "base64")
                .toString("ascii")
                .trim();
            try {
                const result = parser.parseLine(decodedData);
                if (result.request == "") {
                    // empty request - drop message
                    return {
                        eventID: record.eventID,
                        invokeIdentityArn: record.invokeIdentityArn,
                        eventVersion: record.eventVersion,
                        eventName: record.eventName,
                        eventSourceARN: record.eventSourceARN,
                        result: "Dropped"
                    };
                }
                return {
                    // successfully parsed message
                    eventID: record.eventID,
                    invokeIdentityArn: record.invokeIdentityArn,
                    eventVersion: record.eventVersion,
                    eventName: record.eventName,
                    eventSourceARN: record.eventSourceARN,
                    kinesis: {
                        data: new Buffer(JSON.stringify(result)).toString(
                            "base64"
                        ),
                    },
                    result: "Ok"
                };
            } catch (err) {
                // error - fail message
                return {
                    eventID: record.eventID,
                    invokeIdentityArn: record.invokeIdentityArn,
                    eventVersion: record.eventVersion,
                    eventName: record.eventName,
                    eventSourceARN: record.eventSourceARN,
                    result: "ProcessingFailed",
                };
            }
        })
    };
};

(Опционально) Подготовьте файл схемы данных в формате JSON.

Пример файла со схемой данных:

[
    {
        "name": "<имя_поля>",
        "type": "<тип>"
    },
    ...
    {
        "name": "<имя_поля>",
        "type": "<тип>"
    }
]

Список допустимых типов:

any
boolean
datetime
double
int8
int16
int32
int64
string
uint8
uint16
uint32
uint64
utf8

Источник Yandex Managed Service for YDB

Если вы выбрали режим базы данных Dedicated, создайте и настройте группу безопасности в сети, где находится БД.

Подготовка приемника

Приемник ClickHouse®

Managed Service for ClickHouse®

ClickHouse®

Создайте базу-приемник.

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

После старта трансфер подключится к приемнику от имени этого пользователя.
Если в кластере включено управление пользователями через SQL, выдайте созданному пользователю права:
```
GRANT CLUSTER ON *.* TO <имя_пользователя>
GRANT SELECT, INSERT, ALTER DELETE, CREATE TABLE, CREATE VIEW, DROP TABLE, TRUNCATE, dictGet ON <имя_базы_данных>.* TO <имя_пользователя>
GRANT SELECT(macro, substitution) ON system.macros TO <имя_пользователя>
```
Если управление пользователями через SQL выключено, права назначаются через консоль управления и CLI.
Создайте группу безопасности и настройте ее.
Назначьте кластеру Managed Service for ClickHouse® созданную группу безопасности.

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

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

После старта трансфер подключится к приемнику от имени этого пользователя.

Выдайте созданному пользователю права:

GRANT CLUSTER ON *.* TO <имя_пользователя>
GRANT SELECT, INSERT, ALTER DELETE, CREATE TABLE, CREATE VIEW, DROP TABLE, TRUNCATE, dictGet ON <имя_базы_данных>.* TO <имя_пользователя>
GRANT SELECT(macro, substitution) ON system.macros TO <имя_пользователя>

Приемник Greenplum®

Yandex MPP Analytics for PostgreSQL

Greenplum®

Отключите на приемнике следующие настройки:
- проверку целостности внешних ключей;
- триггеры;
- другие ограничения (constraints).
Важно

Не включайте эти настройки до окончания трансфера. Это обеспечит целостность данных по внешним ключам.

Создайте пользователя:

CREATE ROLE <имя_пользователя> LOGIN ENCRYPTED PASSWORD '<пароль>';

Выдайте созданному пользователю все привилегии на базу данных, схемы и переносимые таблицы:
```
GRANT ALL PRIVILEGES ON DATABASE <имя_базы> TO <имя_пользователя>;
```
Если база не пустая, то пользователь должен быть ее владельцем (owner):
```
ALTER DATABASE <имя_базы> OWNER TO <имя_пользователя>;
```
После старта трансфер подключится к приемнику от имени этого пользователя.

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Отключите на приемнике следующие настройки:
- проверку целостности внешних ключей;
- триггеры;
- другие ограничения (constraints).
Важно

Не включайте эти настройки до окончания трансфера. Это обеспечит целостность данных по внешним ключам.

Создайте пользователя:

CREATE ROLE <имя_пользователя> LOGIN ENCRYPTED PASSWORD '<пароль>';

Выдайте созданному пользователю все привилегии на базу данных, схемы и переносимые таблицы:
```
GRANT ALL PRIVILEGES ON DATABASE <имя_базы> TO <имя_пользователя>;
```
Если база не пустая, то пользователь должен быть ее владельцем (owner):
```
ALTER DATABASE <имя_базы> OWNER TO <имя_пользователя>;
```
После старта трансфер подключится к приемнику от имени этого пользователя.

Приемник MongoDB

Yandex StoreDoc (Managed Service for MongoDB)

MongoDB

Создайте базу данных.
Создайте пользователя с ролью readWrite на созданную базу.
Чтобы шардировать переносимые коллекции в кластере-приемнике Yandex StoreDoc:
1. Следуя инструкции, создайте и настройте в базе-приемнике пустые шардированные коллекции.
  
  Сервис Data Transfer не шардирует переносимые коллекции автоматически. Шардирование больших коллекций может занять продолжительное время и снизить скорость трансфера.
2. Если шардирование происходит по ключу, отличному от _id (используется по умолчанию), назначьте пользователю роль mdbShardingManager.
3. При создании эндпоинта для приемника выберите политику очистки DISABLED или TRUNCATE.
  
  Выбор политики DROP приведет к тому, что при активации трансфера сервис удалит из базы-приемника все данные, в т. ч. шардированные коллекции, и создаст вместо них новые, нешардированные.

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Убедитесь, что версия MongoDB на приемнике не ниже, чем на источнике.
Убедитесь, что кластер MongoDB сконфигурирован таким образом, чтобы на запросы к нему он возвращал корректно разрешающиеся IP-адреса или FQDN (Fully Qualified Domain Name).
Настройте кластер-приемник, чтобы к нему можно было подключиться из интернета:
1. Измените в конфигурационном файле значение настройки net.bindIp со 127.0.0.1 на 0.0.0.0:
```
# network interfaces
net:
  port: 27017
  bindIp: 0.0.0.0
```
2. Перезапустите сервис mongod:
```
sudo systemctl restart mongod.service
```
Если кластер-приемник не использует репликацию, включите ее:
1. Добавьте в конфигурационный файл /etc/mongod.conf настройки репликации:
```
replication:
  replSetName: <имя_набора_реплик>
```
2. Перезапустите сервис mongod:
```
sudo systemctl restart mongod.service
```
3. Подключитесь к MongoDB и инициализируйте набор реплик командой:
```
rs.initiate({
    _id: "<имя_набора_реплик>",
    members: [{
        _id: 0,
        host: "<IP-адрес_который_слушает_Yandex_StoreDoc>:<порт>"
    }]
});
```
Подключитесь к кластеру и создайте базу-приемник:
```
use <имя_базы>
```

Создайте пользователя с правами readWrite на базу-приемник:

use admin;
db.createUser({
    user: "<имя_пользователя>",
    pwd: "<пароль>",
    mechanisms: ["SCRAM-SHA-1"],
    roles: [
        {
            db: "<имя_базы-приемника>",
            role: "readWrite"
        }
    ]
});

После старта трансфер подключится к приемнику от имени этого пользователя.

Чтобы шардировать переносимые коллекции в кластере-приемнике:
1. Подготовьте базу данных и создайте в ней пустые коллекции с теми же именами, что и на источнике.
  
  Сервис Data Transfer не шардирует переносимые коллекции автоматически. Шардирование больших коллекций может занять продолжительное время и снизить скорость трансфера.
2. Включите шардирование для базы-приемника:
```
sh.enableSharding("<имя_базы-приемника>")
```
3. Задайте шардирование для каждой коллекции с учетом ее пространства имен:
```
sh.shardCollection("<имя_базы-приемника>.<имя_коллекции>", { <имя_поля>: <1|"hashed">, ... });
```
4. Чтобы убедиться в том, что шардирование настроено и включено, получите список доступных шардов:
```
sh.status()
```
5. Если для шардирования используется ключ, отличный от _id (значение по умолчанию), назначьте системную роль clusterManager пользователю, от имени которого сервис Data Transfer будет подключаться к кластеру-приемнику:
```
use admin;
db.grantRolesToUser("<имя_пользователя>", ["clusterManager"]);
```
6. При создании эндпоинта для приемника выберите политику очистки DISABLED или TRUNCATE.
  
  Выбор политики DROP приведет к тому, что при активации трансфера сервис удалит из базы-приемника все данные, в т. ч. шардированные коллекции, и создаст вместо них новые, нешардированные.

Приемник MySQL®

Managed Service for MySQL®

MySQL®

Убедитесь, что мажорная версия MySQL® на приемнике не ниже версии на источнике.
Установите SQL Mode, который совпадает с источником.
Создайте пользователя для подключения к приемнику.
1. Назначьте пользователю роль ALL_PRIVILEGES для базы-приемника.

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Убедитесь, что мажорная версия MySQL® на приемнике не ниже версии на источнике.
Убедитесь, что приемник использует подсистему хранения данных низкого уровня MyISAM или InnoDB.
Установите SQL Mode, который совпадает с источником.

Создайте пользователя для подключения к приемнику и выдайте ему необходимые привилегии:

CREATE USER '<имя_пользователя>'@'%' IDENTIFIED BY '<пароль>';
GRANT ALL PRIVILEGES ON <имя_базы>.* TO '<имя_пользователя>'@'%';

Приемник Yandex Object Storage

Создайте бакет нужной вам конфигурации.
Создайте сервисный аккаунт с ролью storage.uploader.

Приемник Elasticsearch

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.

Убедитесь, что количество колонок в источнике не превышает максимальное количество полей в индексах Elasticsearch. Максимальное количество полей задается в параметре index.mapping.total_fields.limit и по умолчанию составляет 1000.

Чтобы увеличить значение параметра, настройте шаблон, по которому максимальное количество полей в создаваемых индексах будет равно указанному значению.
Пример запроса для настройки шаблона
```
curl \
    --user <имя_пользователя_Elasticsearch>:<пароль> \
    --header 'Content-Type: application/json' \
    --request PUT "https://<FQDN_кластера_Elasticsearch>:9200/_template/index_defaults" \
    --data '
        {
            "index_patterns": "cdc*",
            "settings": {
                "index": {
                    "mapping": {
                        "total_fields": {
                            "limit": "2000"
                        }
                    }
                }
            }
        }'
```
При такой настройке шаблона, все новые индексы с маской cdc* смогут содержать до 2000 полей.
Для настройки шаблона можно также использовать интерфейс Kibana.

Проверить текущее значение параметра index.mapping.total_fields.limit можно с помощью интерфейса Kibana, либо выполнив запрос:
```
curl \
    --user <имя_пользователя_Elasticsearch>:<пароль> \
    --header 'Content-Type: application/json' \
    --request GET 'https://<FQDN_кластера_Elasticsearch>:9200/<название_индекса>/_settings/*total_fields.limit?include_defaults=true'
```
По умолчанию при трансфере данных в единичный индекс задействуется только один хост. Чтобы распределить нагрузку между хостами при передаче больших объемов данных, настройте шаблон, по которому создаваемые индексы будут заранее разбиты на шарды.
Пример запроса для настройки шаблона
```
curl \
    --user <имя_пользователя_Elasticsearch>:<пароль> \
    --header 'Content-Type: application/json' \
    --request PUT 'https://<FQDN_кластера_Elasticsearch>:9200/_template/index_defaults' \
    --data '
        {
            "index_patterns": "cdc*",
            "settings" : {
                "index" : {
                    "number_of_shards" : 15,
                    "number_of_replicas" : 1
                }
            }
        }'
```
При такой настройке шаблона, все новые индексы с маской cdc* будут разбиты на 15 шардов.

Для настройки шаблона можно также использовать интерфейс Kibana.

Приемник OpenSearch

Managed Service for OpenSearch

OpenSearch

Убедитесь, что количество колонок в источнике не превышает максимальное количество полей в индексах OpenSearch. Максимальное количество полей задается в параметре index.mapping.total_fields.limit и по умолчанию составляет 1000.

Важно

Превышение лимита приведет к ошибке Limit of total fields [1000] has been exceeded и остановке трансфера.

Чтобы увеличить значение параметра, настройте шаблон, по которому максимальное количество полей в создаваемых индексах будет равно указанному значению.
Пример запроса для настройки шаблона
```
curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT "https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_template/index_defaults" \
--data '
    {
        "index_patterns": "cdc*",
        "settings": {
            "index": {
                "mapping": {
                    "total_fields": {
                        "limit": "2000"
                    }
                }
            }
        }
    }'
```
При такой настройке шаблона все новые индексы с маской cdc* смогут содержать до 2000 полей.
Для настройки шаблона можно также использовать интерфейс OpenSearch Dashboards.

Чтобы проверить текущее значение параметра index.mapping.total_fields.limit, выполните запрос:
```
curl \
    --user <имя_пользователя_OpenSearch>:<пароль> \
    --header 'Content-Type: application/json' \
    --request GET 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/<название_индекса>/_settings/*total_fields.limit?include_defaults=true'
```
По умолчанию при трансфере данных в единичный индекс задействуется только один хост. Чтобы распределить нагрузку между хостами при передаче больших объемов данных, настройте шаблон, по которому создаваемые индексы будут заранее разбиты на шарды.
Пример запроса для настройки шаблона
```
curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_template/index_defaults' \
--data '
    {
        "index_patterns": "cdc*",
        "settings" : {
            "index" : {
                "number_of_shards" : 15,
                "number_of_replicas" : 1
            }
        }
    }'
```
При такой настройке шаблона, все новые индексы с маской cdc* будут разбиты на 15 шардов.

Для настройки шаблона можно также использовать интерфейс OpenSearch Dashboards.

Чтобы повысить безопасность и доступность данных, установите политику, которая будет создавать новый индекс при выполнении хотя бы одного из условий (рекомендуемые значения):

Когда размер индекса превысит 50 ГБ.
Когда возраст индекса превысит 30 дней.

Создать и включить политику можно с помощью запросов. Подробнее о политиках см. в документации OpenSearch.

Пример запроса для создания политики

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_plugins/_ism/policies/rollover_policy' \
--data '
    {
        "policy": {
            "description": "Example rollover policy",
            "default_state": "rollover",
            "schema_version": 1,
            "states": [
                {
                    "name": "rollover",
                    "actions": [
                        {
                            "rollover": {
                                "min_index_age": "30d",
                                "min_primary_shard_size": "50gb"
                            }
                        }
                    ],
                    "transitions": []
                }
            ],
            "ism_template": {
                "index_patterns": ["log*"],
                "priority": 100
            }
        }
    }'

Пример запроса для назначения политике псевдонима

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_index_template/ism_rollover' \
--data '
    {
        "index_patterns": ["log*"],
        "template": {
            "settings": {
                "plugins.index_state_management.rollover_alias": "log"
            }
        }
    }'

Пример запроса для создания индекса с псевдонимом политики

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/log-000001' \
--data '
    {
        "aliases": {
            "log": {
                "is_write_index": true
            }
        }
    }'

Пример запроса для проверки, прикреплена ли политика к индексу

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request GET 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_plugins/_ism/explain/log-000001?pretty'

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.

Убедитесь, что количество колонок в источнике не превышает максимальное количество полей в индексах OpenSearch. Максимальное количество полей задается в параметре index.mapping.total_fields.limit и по умолчанию составляет 1000.

Важно

Превышение лимита приведет к ошибке Limit of total fields [1000] has been exceeded и остановке трансфера.

Чтобы увеличить значение параметра, настройте шаблон, по которому максимальное количество полей в создаваемых индексах будет равно указанному значению.
Пример запроса для настройки шаблона
```
curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT "https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_template/index_defaults" \
--data '
    {
        "index_patterns": "cdc*",
        "settings": {
            "index": {
                "mapping": {
                    "total_fields": {
                        "limit": "2000"
                    }
                }
            }
        }
    }'
```
При такой настройке шаблона все новые индексы с маской cdc* смогут содержать до 2000 полей.
Для настройки шаблона можно также использовать интерфейс OpenSearch Dashboards.

Чтобы проверить текущее значение параметра index.mapping.total_fields.limit, выполните запрос:
```
curl \
    --user <имя_пользователя_OpenSearch>:<пароль> \
    --header 'Content-Type: application/json' \
    --request GET 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/<название_индекса>/_settings/*total_fields.limit?include_defaults=true'
```
По умолчанию при трансфере данных в единичный индекс задействуется только один хост. Чтобы распределить нагрузку между хостами при передаче больших объемов данных, настройте шаблон, по которому создаваемые индексы будут заранее разбиты на шарды.
Пример запроса для настройки шаблона
```
curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_template/index_defaults' \
--data '
    {
        "index_patterns": "cdc*",
        "settings" : {
            "index" : {
                "number_of_shards" : 15,
                "number_of_replicas" : 1
            }
        }
    }'
```
При такой настройке шаблона, все новые индексы с маской cdc* будут разбиты на 15 шардов.

Для настройки шаблона можно также использовать интерфейс OpenSearch Dashboards.

Когда размер индекса превысит 50 ГБ.
Когда возраст индекса превысит 30 дней.

Создать и включить политику можно с помощью запросов. Подробнее о политиках см. в документации OpenSearch.

Пример запроса для создания политики

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_plugins/_ism/policies/rollover_policy' \
--data '
    {
        "policy": {
            "description": "Example rollover policy",
            "default_state": "rollover",
            "schema_version": 1,
            "states": [
                {
                    "name": "rollover",
                    "actions": [
                        {
                            "rollover": {
                                "min_index_age": "30d",
                                "min_primary_shard_size": "50gb"
                            }
                        }
                    ],
                    "transitions": []
                }
            ],
            "ism_template": {
                "index_patterns": ["log*"],
                "priority": 100
            }
        }
    }'

Пример запроса для назначения политике псевдонима

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_index_template/ism_rollover' \
--data '
    {
        "index_patterns": ["log*"],
        "template": {
            "settings": {
                "plugins.index_state_management.rollover_alias": "log"
            }
        }
    }'

Пример запроса для создания индекса с псевдонимом политики

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request PUT 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/log-000001' \
--data '
    {
        "aliases": {
            "log": {
                "is_write_index": true
            }
        }
    }'

Пример запроса для проверки, прикреплена ли политика к индексу

curl \
--user <имя_пользователя_OpenSearch>:<пароль> \
--header 'Content-Type: application/json' \
--request GET 'https://<адрес_хоста_OpenSearch_с_ролью_DATA>:9200/_plugins/_ism/explain/log-000001?pretty'

Приемник PostgreSQL

Managed Service for PostgreSQL

PostgreSQL

Убедитесь, что мажорная версия PostgreSQL на приемнике не ниже версии на источнике.
При трансфере из PostgreSQL включите те же расширения в базе приемника, что и в базе источника.

Если в базе источника расширения установлены в пользовательскую схему, и эти расширения используются в DDL переносимых объектов, то вручную создайте в приемнике DDL. В этих DDL обращение к функции должно быть без указания схемы. В политике очистки эндпоинта-приемника установите значение Truncate, чтобы трансфер не удалил эти объекты.
Выберите политику очистки таблиц трансфера Drop.

Если вы вручную создали в приемнике DDL, используйте политику Truncate. При политике Truncate эти DDL не будут удалены.
Создайте пользователя с доступом к базе приемника.
Выдайте созданному пользователю все привилегии на базу данных, схемы и переносимые таблицы:
```
GRANT ALL PRIVILEGES ON DATABASE <имя_базы> TO <имя_пользователя>;
```
Если база не пустая, то пользователь должен быть ее владельцем (owner):
```
ALTER DATABASE <имя_базы> OWNER TO <имя_пользователя>;
```
После старта трансфер подключится к приемнику от имени этого пользователя.
Если в приемнике включена опция сохранение границ транзакций, выдайте созданному пользователю все привилегии на создание служебной таблицы __data_transfer_lsn в текущей схеме (обычно public) на приемнике:
```
GRANT ALL PRIVILEGES ON SCHEMA <имя_схемы> TO <имя_пользователя>;
```
Настройте количество подключений пользователя к базе данных.

Если вы не планируете использовать для подключения к внешнему кластеру сервис Cloud Interconnect или VPN, разрешите подключения к такому кластеру из интернета с IP-адресов, используемых сервисом Data Transfer.

Подробнее о настройке сети для работы с внешними ресурсами см. в концепции.
Убедитесь, что мажорная версия PostgreSQL на приемнике не ниже версии на источнике.
Включите те же расширения в базе приемника, что и в базе источника.
Убедитесь, что на приемнике выбрана политика очистки DROP таблиц трансфера.

Создайте пользователя:

CREATE ROLE <имя_пользователя> LOGIN ENCRYPTED PASSWORD '<пароль>';

Выдайте созданному пользователю все привилегии на базу данных, схемы и переносимые таблицы:
```
GRANT ALL PRIVILEGES ON DATABASE <имя_базы> TO <имя_пользователя>;
```
Если база не пустая, то пользователь должен быть ее владельцем (owner):
```
ALTER DATABASE <имя_базы> OWNER TO <имя_пользователя>;
```
После старта трансфер подключится к приемнику от имени этого пользователя.
Если в приемнике включена опция сохранение границ транзакций, выдайте созданному пользователю все привилегии на создание служебной таблицы __data_transfer_lsn в текущей схеме (обычно public) на приемнике:
```
GRANT ALL PRIVILEGES ON SCHEMA <имя_схемы> TO <имя_пользователя>;
```
Настройте количество подключений пользователя к базе данных.

Приемник Yandex Managed Service for YDB

Создайте сервисный аккаунт с ролью ydb.editor.
Для базы данных в Dedicated-режиме создайте и настройте группу безопасности в сети, где находится БД.

Airbyte® является зарегистрированным товарным знаком Airbyte, Inc в США и/или других странах.

Greenplum® и Greenplum Database® являются зарегистрированными товарными знаками или товарными знаками Broadcom Inc в США и/или других странах.

ClickHouse® является зарегистрированным товарным знаком ClickHouse, Inc.

Подготовка к трансферу

Подготовка источникаПодготовка источника

Источники Airbyte®Источники Airbyte®

Источник AWS CloudTrailИсточник AWS CloudTrail

Источник BigQueryИсточник BigQuery

Источник Microsoft SQL ServerИсточник Microsoft SQL Server

Источник S3Источник S3

Источник Apache Kafka®Источник Apache Kafka®

Источник ClickHouse®Источник ClickHouse®

Источник Greenplum®Источник Greenplum®

Источник MongoDBИсточник MongoDB

Источник MySQL®Источник MySQL®

Источник ElasticsearchИсточник Elasticsearch

Источник OpenSearchИсточник OpenSearch

Источник OracleИсточник Oracle

Источник PostgreSQLИсточник PostgreSQL

Источник Yandex Data StreamsИсточник Yandex Data Streams

Источник Yandex Managed Service for YDBИсточник Yandex Managed Service for YDB

Подготовка приемникаПодготовка приемника

Приемник ClickHouse®Приемник ClickHouse®

Приемник Greenplum®Приемник Greenplum®

Приемник MongoDBПриемник MongoDB

Приемник MySQL®Приемник MySQL®

Приемник Yandex Object StorageПриемник Yandex Object Storage

Приемник ElasticsearchПриемник Elasticsearch

Приемник OpenSearchПриемник OpenSearch

Приемник PostgreSQLПриемник PostgreSQL

Приемник Yandex Managed Service for YDBПриемник Yandex Managed Service for YDB

Была ли статья полезна?

Подготовка источника

Источники Airbyte®

Источник AWS CloudTrail

Источник BigQuery

Источник Microsoft SQL Server

Источник S3

Источник Apache Kafka®

Источник ClickHouse®

Источник Greenplum®

Источник MongoDB

Источник MySQL®

Источник Elasticsearch

Источник OpenSearch

Источник Oracle

Источник PostgreSQL

Источник Yandex Data Streams

Источник Yandex Managed Service for YDB

Подготовка приемника

Приемник ClickHouse®

Приемник Greenplum®

Приемник MongoDB

Приемник MySQL®

Приемник Yandex Object Storage

Приемник Elasticsearch

Приемник OpenSearch

Приемник PostgreSQL

Приемник Yandex Managed Service for YDB