Только хардкор, только мануал: репликация данных

Привет, Хабр!

Каждая компания стремится к тому, чтобы данные были не только доступны в нужный момент, но и надежно защищены (спасибо, КЭП!). Более того, необходимость обеспечения безопасности ЗОКИИ и требований по импортозамещению ставит новые задачи в области интеграции систем и миграции данных. Один из способов разобраться с этими вызовами — репликация данных. Она помогает компаниям справляться с растущими нагрузками, обеспечивает защиту данных и облегчает миграцию между различными платформами. Однако ее успешное внедрение требует тщательной подготовки, внимательного выбора стратегии и регулярного мониторинга. Хватит слов – за дело!  В этой статье мы покажем сразу несколько технических решений.

Репликация данных между Oracle и PostgreSQL с помощью Oracle GoldenGate

Начнем с самого актуального на данный момент кейса репликации Oracle -> Postgres.

Сначала готовим сервер-источник (Oracle). Помним, база должна быть в режиме Archive Logging — это позволит нам отслеживать данные.

Подключаемся к Oracle через Sqlplus, поднимаем БД в mount и делаем следующие действия:

— включаем архивацию логов:

SQL> ALTER DATABASE ARCHIVELOG; ALTER DATABASE OPEN;

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

SQL> ALTER DATABASE FORCE LOGGING;

— включаем дополнительное логирование:

SQL> ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

— проверяем:

SQL> SELECT log_mode, supplemental_log_data_min, force_logging FROM v$database;

— создаем в БД пользователя для GoldenGate:

SQL> CREATE USER ggadmin IDENTIFIED BY ggpassword DEFAULT TABLESPACE users TEMPORARY TABLESPACE temp;

— выдаем привилегии для работы с GoldenGate

SQL> GRANT CONNECT, RESOURCE TO ggadmin; GRANT CREATE SESSION TO ggadmin; GRANT SELECT ANY DICTIONARY TO ggadmin; GRANT EXECUTE ON DBMS_LOGMNR TO ggadmin; GRANT EXECUTE ON DBMS_LOGMNR_D TO ggadmin; GRANT EXECUTE ON DBMS_FLASHBACK TO ggadmin; GRANT SELECT ON V_$DATABASE TO ggadmin; GRANT SELECT ON ALL_TABLES TO ggadmin; GRANT SELECT ON DBA_REGISTERED_ARCHIVED_LOG TO ggadmin; GRANT SELECT ON DBA_CAPTURE TO ggadmin; GRANT SELECT ON DBA_APPLY TO ggadmin;

— разрешаем пользователю GGADMIN использовать репликацию GoldenGate:

SQL> EXEC dbms_goldengate_auth.grant_admin_privilege('GGADMIN'); ALTER SYSTEM SET ENABLE_GOLDENGATE_REPLICATION=TRUE SCOPE=BOTH;

— настраиваем логирование на уровне реплицируемой таблицы:

SQL> ALTER TABLE schema_name.table_name ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS.

Теперь переходим к установке и настройке GoldenGate Extract на сервере Oracle.

Для начала скачиваем cам GG. Перемещаем архив на целевой хост с нашей базой-источником и разархивируем его:

bash> mkdir -p /u01/app/ogg  mv ~/ggs_*.zip /u01/app/ogg  cd /u01/app/ogg  unzip ggs_*.zip

Запускаем ggsci:

bash> cd /u01/app/ogg ./ggsci ./ggsci

Создаем каталоги:

GGSCI> CREATE SUBDIRS

Рекомендуем добавить OGG в PATH — так удобнее запускать команды:

bash> echo "export OGG_HOME=/u01/app/ogg" >> ~/.bashrc echo "export PATH=\$OGG_HOME:\$PATH" >> ~/.bashrc source ~/.bashrc

Теперь настраиваем параметры процесс-менеджера. Он нам нужен для управления процессами, которые отслеживают изменения и передают их в Postgres:

GGSCI> EDIT PARAM MGR

Добавляем его:

PORT 7809 DYNAMICPORTLIST 7810-7820 ACCESSRULE, PROG *, IPADDR *, ALLOW

А теперь запускаем:

GGSCI> START MGR

Задаем учетные данные для подключения к БД-источнику:

GGSCI> ADD CREDENTIALSTORE ALTER CREDENTIALSTORE ADD USER ggadmin@ORACLE_TNS_ALIAS ALIAS ggadmin

Настраиваем сам процесс Extract:

GGSCI> DBLOGIN USERIDALIAS ggadmin ADD EXTRACT eORACLE TRANLOG, INTEGRATED TRANLOG, BEGIN NOW ADD EXTTRAIL ./dirdat/lt, EXTRACT eORACLE REGISTER EXTRACT eORACLE DATABASE

Редактируем его параметры:

GGSCI> EDIT PARAM eORACLE

Добавляем процесс экстрактора:

EXTRACT eORACLE USERIDALIAS ggadmin TRANLOGOPTIONS INTEGRATEDPARAMS (DOWNSTREAM_REAL_TIME_MINE Y) EXTTRAIL ./dirdat/lt TABLE schema_name.table_name;

Запускаем его:

GGSCI> START eORACLE

Теперь настраиваем процесс, который передает изменения по сети, — PUMP:

GGSCI> ADD EXTRACT pORACLE EXTTRAILSOURCE ./dirdat/lt ADD RMTTRAIL ./dirdat/rt, EXTRACT pORACLE

Редактируем параметры:

GGSCI> EDIT PARAM pORACLE

Добавляем процесс пампа:

EXTRACT pORACLE USERIDALIAS ggadmin RMTHOST target_host, MGRPORT 7810 RMTTRAIL ./dirdat/rt PASSTHRU TABLE schema_name.table_name;

Запускаем его:

GGSCI> START pORACLE

Теперь переходим к серверу-приемнику — PostgreSQL. Создаем пользователя и выдаем ему права.

Подключаемся к PostgreSQL при помощи команды psql -U postgres и переходим в нужную БД \c dbname

Затем выполняем:

CREATE USER ggadmin WITH LOGIN PASSWORD 'ggpassword'; GRANT CONNECT ON DATABASE target_db TO ggadmin; GRANT USAGE ON SCHEMA target_schema TO ggadmin; GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON ALL TABLES IN SCHEMA target_schema TO ggadmin;

Автоматически выдаем права на будущие таблицы:

ALTER DEFAULT PRIVILEGES IN SCHEMA target_schema GRANT SELECT, INSERT, UPDATE, DELETE, TRUNCATE ON TABLES TO ggadmin;

Устанавливаем GoldenGate по аналогии, как делали с сервером ORACLE, и настраиваем ODBC. Редактируем odbc.ini:

[ODBC Data Sources] GG_Postgres=DataDirect 9.6 PostgreSQL Wire Protocol  [GG_Postgres] Driver=/ogg/lib/ggpsql25.so Description=PostgreSQL Wire Protocol Database=target_db HostName=localhost PortNumber=5432 LogonID=ggadmin Password=ggpassword

Отлично, подготовка выполнена. Теперь устанавливаем и переходим к настройке Replicat на сервере PostgreSQL.

Первым делом создаем каталоги:

GGSCI> CREATE SUBDIRS

Редактируем параметры менеджера:

GGSCI> EDIT PARAM MGR PORT 7810 DYNAMICPORTLIST 7811-7820 ACCESSRULE, PROG REPLICAT, IPADDR *, ALLOW

Запускаем менеджер:

GGSCI> START MGR

Добавляем процесс Replicat:

GGSCI> DBLOGIN USERID ggadmin PASSWORD ggpassword ADD REPLICAT rPOSTGRES EXTTRAIL ./dirdat/rt, CHECKPOINTTABLE ggadmin.checkpoint

Редактируем параметры:

GGSCI> EDIT PARAM rPOSTGRES REPLICAT rPOSTGRES SETENV (PGCLIENTENCODING = "UTF8") SETENV (ODBCINI="/ogg/odbc.ini") SETENV (NLS_LANG="AMERICAN_AMERICA.AL32UTF8") TARGETDB GG_Postgres, USERID ggadmin, PASSWORD ggpassword MAP schema_name.table_name, TARGET target_schema.table_name;

Запускаем репликат:

GGSCI> START rPOSTGRES

Проверяем состояние:

GGSCI> INFO ALL

Теперь нам нужно убедиться, что все работает. На стороне Oracle вставляем строку в таблицу:

SQL> INSERT INTO schema_name.table_name VALUES (...); COMMIT;

На стороне PostgreSQL проверяем результат:

SELECT * FROM target_schema.table_name;

Если данные появились — репликация работает. Контролировать процесс работы репликатора можно с помощью команды GGSCI> VIEW REPORT rPOSTGRES

Good job!

Репликация данных PostgreSQL→ORACLE с помощью Oracle GoldenGate

Для репликации на первом этапе настраиваем средство — Oracle GoldenGate.

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

Распаковываем архив с файлами Oracle GoldenGate в выбранный каталог.

Теперь настраиваем Oracle GoldenGate. Для этого переходим в домашний каталог Oracle GoldenGate, запускаем консоль управления GGSCI (./ggsci) и создаем поддиректории:

GGSCI> create subdirs

В настройках MANAGER прописываем порт, который он будет слушать:

GGSCI> edit param MGR PORT 7800 GGSCI> start MGR

Создаем файл GLOBALS и добавляем схему ggs, где будут храниться технические данные GoldenGate со стороны PostgreSQL. Файл Globals содержит в себе глобальные настройки, которые влияют на работу всей инсталляции:

GGSCI> edit param ./GLOBALS GGSCHEMA ggadmn

Теперь подготавливаем PostgreSQL для настройки репликации OGG. Включаем журналирование WAL в файле конфигурации PostgreSQL postgresql.conf: wal_level = logical.

Примечание автора: Этот параметр устанавливает уровень логирования WAL (Write-Ahead Logging) на «logical», который необходим для логической репликации. Она передает изменения данных через логические слоты. Таким образом, в WAL-файлы начинают попадать дополнительные данные, которые используются процессами репликации.

Теперь включаем параметр max_replication_slots для поддержки репликации: max_replication_slots = 5.

Он определяет максимальное количество слотов репликации, которые PostgreSQL может поддерживать одновременно. Это необходимо для работы GoldenGate, который использует слот для отслеживания изменений. Остальные слоты могут использоваться для физической репликации и бэкапов, поэтому их количество нужно увеличивать.

Перезапускаем PostgreSQL для применения изменений. Сначала входим в PostgreSQL: psql -U postgres. Подключаемся к базе Postgres от его системного пользователя.

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

CREATE ROLE ggadmn LOGIN PASSWORD ‘password’; Создаем пользователя ggadmn с паролем.

Назначаем права репликации: 

ALTER ROLE ggadmn REPLICATION; 

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

Теперь создаем пользователя для репликации на приемнике (Oracle БД).

Для этого выполняем:

SQL>create user ggadmn identified by ggadmn default tablespace users temporary SQL>tablespace temp; -- Создает пользователя в Oracle с указанным паролем и таблицами по умолчанию.

SQL>grant connect to  ggadmn; -- Разрешает пользователю подключаться к базе данных.

SQL>grant resource to  ggadmn; -- Предоставляет базовые права для работы с объектами базы данных.

SQL>exec dbms_goldengate_auth.grant_admin_privilege ('ggadmn'); --Назначает административные привилегии для GoldenGate.

SQL>alter system set enable_goldengate_replication=TRUE scope=both; --включает поддержку репликации GoldenGate.

Переходим к настройке GoldenGate со стороны источника (PostgreSQL). Создаем файл параметров для источника (PostgreSQL). Для этого настраиваем odbc.ini исходя из своих локальных настроек. Затем создаем файл параметров для PostgreSQL:

bash> vi dirprm/extname.prm

Добавляем в него следующие строки:

EXTRACT extname --Имя Extract-процесса.

SOURCEDB dsn_name USERIDALIAS alias --Подключение к базе PostgreSQL через DSN и алиас пользователя.

RMTHOST <TARGET_IP_ADDRESS>, MGRPORT 7809 --казывает IP-адрес и порт менеджера GoldenGate на стороне Oracle.

RMTTRAIL /u01/app/oracle/product/gg/dirdat/pg --указывает путь для временных файлов данных (trail-файлы).

GETTRUNCATES --Включает обработку команд TRUNCATE.

TABLE schema.object; --Указывает, какие таблицы должны быть реплицированы.

Теперь нам необходимо подключиться к БД-источнику при помощи утилиты GGSCI:

GGSCI>

DBLOGIN SOURCEDB dsn USERIDALIAS alias --Устанавливает соединение с PostgreSQL для GoldenGate.

Далее регистрируем Extract с помощью команды GGSCI. Она создает слот репликации. Помним, что имя extract не может содержать более восьми буквенно-цифровых символов:

GGSCI>

REGISTER EXTRACT extname --Создает слот репликации для логической репликации PostgreSQL.

Делаем Extract и соответствующий ему trail-файл:

GGSCI>

ADD EXTRACT extname, TRANLOG, BEGIN NOW --Создает Extract-процесс для получения данных из транзакционного лога.

GGSCI>

ADD RMTTRAIL /u01/app/oracle/product/gg/dirdat/pg, extract extname --Указывает, где сохранять trail-файлы.

Включаем расширенное логирование для таблицы-источника:

ADD TRANDATA schema.object --Включает расширенное логирование для указанных таблиц, чтобы отслеживать изменения на уровне строк.

Теперь принимаемся за Golden Gate.

Переключаемся на настройку GoldenGate со стороны источника (Oracle).

Добавляем checkpoint table:

GGSCI>

add checkpointtable ggadmn.repchkpt -- Создает таблицу контрольных точек для синхронизации репликации.

Создаем Replicat:

GGSCI>

add replicat ORAREP, exttrail /u01/app/oracle/product/gg/dirdat/pg --Создает процесс Replicat, который считывает trail-файлы и применяет изменения в Oracle.

GGSCI>

edit param ORAREP

REPLICAT ORAREP --Имя процесса Replicat.

USERID ggadmn, PASSWORD ggadmn --Данные для подключения к Oracle.

MAP schema.object, TARGET schema.object;-- Указывает сопоставление схем и таблиц между источником и приемником.

Запускаем Extract и Replicat:

GGSCI>

start extract extname

start replicat ORAREP

Важно: типы полей в Оracle и PostgreSQL должны совпадать и поддерживаться GoldenGate. Подробнее можно прочитать здесь.

Репликация данных Oracle→CockroachDB

Начинаем с переноса структуры таблиц в CockroachDB. На ноде CockroachDB (sourceDB) в БД создаем пользователя ggadmin (нужен для репликата) и выдаем грант admin.

Выпускаем клиентский сертификат для ggadmin:

cockroach cert create-client ggadmin --certs-dir=/cockroach/cert/ --ca-key=/cockroach/cert/ca.key 

Проверяем подключение к БД под УЗ ggadmin при помощи команды:

cockroach sql --certs-dir=/var/lib/cockroach/certs --user=ogg 

Теперь создаем пользователя oracle и назначаем группу cockroach. Устанавливаем GoldenGate for PostgreSQL и создаем структуру каталогов для GG командой create subdirs. Настраиваем и запускаем менеджера GG:

 PORT 7800      DYNAMICPORTLIST 7800-7810      ACCESSRULE, PROG REPLICAT, IPADDR *, ALLOW 

Настраиваем odbc.ini в $GG_HOME:

[ODBC Data Sources]     PostgreSQL on pgsql     [ODBC]     IANAAppCodePage=4     InstallDir=/cockroach/ogg     [cockdb]     Driver=/cockroach/ogg/lib/GGpsql25.so     Description=DataDirect 7.1 PostgreSQL Wire Protocol     Database={dbname}     HostName=localhost     PortNumber=26257     LogonID=ggadmin     Password={password}     EncryptionMethod=1     ValidateServerCertificate=1     TrustStore=/cockroach/cert/ca.crtДелаем репликат для таблицы TABLE_NAME, но не запускаем (сделаем позже):

Делаем репликат для таблицы TABLE_NAME, но не запускаем (сделаем позже):

GGSCI> edit param REP1     REPLICAT REP1     SETENV ( PGCLIENTENCODING = "UTF8" )     SETENV (ODBCINI="/cockroach/ogg/odbc.ini" )     SETENV (NLS_LANG="AMERICAN_AMERICA.AL32UTF8")     TARGETDB {dbname}     DISCARDFILE ./dirrpt/rep1.dsc, purge, MEGABYTES 1024     MAP owner.table, TARGET public.table;

Добавляем репликат:

GGSCI> add replicat REP1, NODBCHECKPOINT, exttrail ./dirdat/ee

На даунстриме сервера downstream_hostname настраиваем памп для трейл-файлов экстракта:

GGSCI> edit param ENAME      EXTRACT ENAME     USERIDALIAS ggadmin     RMTHOST REMOTE_HOST, MGRPORT 7800 TCPBUFSIZE 200000000, TCPFLUSHBYTES 200000000     RMTTRAIL ./dirdat/ee     PASSTHRU     TABLE TABLE_NAME;

Добавляем памп:

GGSCI> Add extract PNAME, EXTTRAILSOURCE ./dirdat/ee Add rmttrail /cockroach/ogg/dirdat/ee, extract PNAME

Запускаем процесс пампа и проверяем, что он запустился и начал отправлять файлы на ноду CockroachDB:

GGSCI> start PNAME

Конфигурируем экстракты для Initial load. Для таблицы TABLE_NAME:

GGSCI> edit param INAME     EXTRACT INAME     SOURCEDB SOURCEDBNAME USERIDALIAS ggadmin     RMTHOST rmthost, mgrport 7800     RMTTASK REPLICAT, GROUP INITLR     TABLE TABLE_NAME;

Добавляем экстракт:

GGSCI> ADD EXTRACT INAME, SOURCEISTABLE

На sourceDB останавливаем накат логов. Снимаем и запоминаем/записываем SCN. Мы же не хотим получить неконсистентные данные. Поэтому требование обязательное.

На даунстриме запускаем процессы Initial load: 

GGSCI> start INAME

И наблюдаем за нагрузкой на CockroachDB.

После завершения первоначальной заливки, если требуется, создаем индексы, Primary Key (очень желательно, потому что его отсутствие негативно влияет на производительность) 
и проверяем статистику по таблицам. Если нужно, пересобираем ее.

На ноде с DESTDB запускаем репликаты с запомненного SCN:

GGSCI> start replicat RNAME, aftercsn <scn_number>

Репликация данных MSSQL→ORACLE с помощью Oracle GoldenGate CDC

Загружаем необходимый дистрибутив GoldenGate for SQL Server, распаковываем его на диск и инсталлируем. Подробную инструкцию по инсталляции читаем на сайте вендора. Для подключения Oracle GoldenGate к SQL Server необходим DSN. Для этого настраиваем  ODBC Data Sources (64-bit).

Нажимаем на Add и выбираем ODBC Driver for SQL Server для вашей версии:

Вводим имя, которое потом будет прописываться для подключения в настройках. Указываем Server:

Далее необходимо настроить параметры подключения:

В следующем окне выбираем БД, к которой будут подключаться:

Далее завершаем настройку. По итогу мастер настройки покажет окно с результатом, там же можно протестировать подключение по DSN:

Теперь переходим к настройке Oracle GoldenGate. Открываем директорию с распакованным дистрибутивом GoldenGate, запускаем консоль управления GGSCI и создаем поддиректории:

GGSCI> Create subdirs

В настройках MANAGER прописываем порт, который он будет слушать:

GGSCI> edit param MGR     PORT 7800 GGSCI> start MGR

Создаем файл GLOBALS и добавляем схему ggs, в которой будут храниться технические данные GoldenGate со стороны SQL Server:

GGSCI> edit param ./GLOBALS     GGSCHEMA ggs

Включаем расширенное логирование для таблицы со стороны MS SQL, участвующей в репликации:

GGSCI> dblogin sourcedb GG add trandata hr.source_table

Теперь начинаем первоначальную загрузку данных таблицы из SQL Server в Oracle (первичная инициализация). Если в таблице источника есть данные и необходимость перенести эти данные на приемник. Копируем их в целевую базу данных с помощью штатного инструмента. В GoldenGate можно использовать функцию Initial Load для предварительной загрузки данных.

Перед проведением работ в файл параметров mgr добавляем строку:

ACCESSRULE, PROG *, IPADDR *, ALLOW

На источнике SQL Server:

GGSCI> ADD EXTRACT INITEX, SOURCEISTABLE EDIT PARAMS INITEX     EXTRACT INITEX     SOURCEDB GG     RMTHOST <TARGET_IP_ADDRESS>, MGRPORT 7809     RMTTASK REPLICAT, GROUP INITRE     TABLE hr.source_table;

На приемнике:

GGSCI> ADD REPLICAT INITRE, SPECIALRUN EDIT PARAMS INITRE     REPLICAT INITRE     USERID ogg, PASSWORD ogg     MAP hr.source_table, TARGET hr.target_table;

После запускаем INITEX. На приемнике INITRE это происходит автоматически:

GGSCI> START INITEX

Теперь настраиваем Extract и Replicat.

На источнике SQL Server:

GGSCI> ADD EXTRACT EXTSQL, TRANLOG, BEGIN NOW add exttrail E:\gg\dirdat\ms, extract EXTSQL edit param EXTSQL EXTRACT EXTSQL SOURCEDB GG EXTTRAIL E:\gg\dirdat\ms TABLE hr.source_table;

Добавляем процесс PUMP:

GGSCI> Add extract DPSQL, EXTTRAILSOURCE E:\gg\dirdat\ms Add rmttrail /u01/app/oracle/product/gg/dirdat/ms, extract DPSQL edit param DPSQL     EXTRACT DPSQL     SOURCEDB GG     RMTHOST <TARGET_IP_ADDRESS>, MGRPORT 7809     RMTTRAIL /u01/app/oracle/product/gg/dirdat/ms     TABLE hr.source_table;

Запускаем Extract:

GGSCI> start EXTSQL

На приемнике Oracle добавляем checkpoint table:

GGSCI> add checkpointtable ogg.repchkpt

Создаем Replicat:

GGSCI> add replicat ORAREP, exttrail /u01/app/oracle/gg/dirdat/ms edit param ORAREP     REPLICAT ORAREP     USERID ogg, PASSWORD ogg     MAP hr.source_table, TARGET hr.target_table;

Запускаем Replicat:

GGSCI> start ORAREP

Теперь создаем задания по очистке данных. Со стороны SQL Server для расширенного логирования данных в таблице используется функция Change Data Capture (CDC). При его включении GoldenGate сам включает CDC и создает задания по захвату и очистке данных. Но стандартное задание по очистке не подходит для работы GoldenGate, поэтому в дистрибутиве есть штатный скрипт по созданию задания, учитывающий функционал GoldenGate.

Именно поэтому для начала необходимо удалить стандартное задание из CDC.

Подключаемся к SQL Server c административными правами и выполняем команду в БД, в которой находятся таблицы источника:

EXEC sys.sp_cdc_drop_job 'cleanup'

Запускаем CMD от имени администратора, переходим в директорию с GoldenGate и выполняем:

ogg_cdc_cleanup_setup.bat createJob user1 pword1 dbname1 HOSTNAME\INSTANCE oggschema

Репликацируем в SQL Server. Для этого из других источников в SQL Server концептуально процесс остается аналогичным.

Настраиваем Replicat:

GGSCI>  EDIT PARAMS cdcrep     REPLICAT cdcrep     TARGETDB targetdsn USERID username PASSWORD password     MAP dbo.*, TARGET dbo.*

Создаем схему для объектов GoldenGate:

CREATE SCHEMA [ogg];

А затем создаем таблицу чекпоинта и добавляем Replicat:

GGSCI> DBLOGIN SOURCEDB targetdsn USERID username PASSWORD password ADD CHECKPOINTTABLE ogg.ggcheck ADD REPLICAT cdcrep, EXTTRAIL ./dirdat/ce, CHECKPOINTTABLE ogg.ggcheck

Запускаем Replicat:

GGSCI> START REPLICAT cdcrep INFO ALL

Репликация данных Oracle→Kafka

В этом разделе мы не будем описывать настройку источника, так как мы описали этот процесс выше.

Устанавливаем GoldenGate 19 for BigData. Для работы потребуется Java — устанавливаем командой: [root@server01 ~]# yum install java-1.8.0-openjdk.x86_64

Добавляем переменные окружения:

bash> export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.275.b01-0.el7_9.x86_64/jre export PATH=$JAVA_HOME/bin:$PATH export LD_LIBRARY_PATH=$JAVA_HOME/lib/amd64/server:$LD_LIBRARY_PATH

Распаковываем дистрибутив GoldenGate for BigData:

bash> mkdir -p /u01/app/oracle/product/19.1.0/oggbigdata_1  cd /u01/app/oracle/product/19.1.0/oggbigdata_1  cp /u01/V983760-01.zip .  unzip V983760-01.zip tar xvf OGG_BigData_Linux_x64_19.1.0.0.1.tar

Для удобства быстрого доступа прописываем Alias:

bash> alias gg='cd /u01/app/oracle/product/19.1.0/oggbigdata_1; ./ggsci'

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

bash> gg  GGSCI>  create subdirs

Добавляем порт в файл параметра менеджера:

GGSCI> edit params mgr     port 7801 GGSCI> start mgr

Настраиваем Kafka. У нас уже установлена и настроена Kafka, но если у вас нет, ниже описываем простой способ развертывания и запуска Kafka:

bash> tar -xzf /tmp/kafka_2.13-2.7.0.tgz cd kafka_2.13-2.7.0/ nohup /home/kafka/kafka_2.13-2.7.0/bin/zookeeper-server-start.sh config/zookeeper.properties > zookeeper.log  nohup /home/kafka/kafka_2.13-2.7.0/bin/kafka-server-start.sh config/server.properties > broker_service.log &

Обычно Kafka располагается на отдельном сервере. В данном случае для работы GoldenGate потребуются библиотеки из Kafka. Ищем с помощью <kafka_install_dir>/libs/*. Эту директорию со всеми файлами копируем на сервере, где будет располагаться GoldenGate. В дальнейшем она понадобится для настроек.

Теперь настраиваем Handler. Первым делом смотрим в директорию AdapterExamples. Она расположена внутри дистрибутива GoldenGate for BigData. В данном случае нас интересует подпапка AdapterExamples/big-data/kafka:

Содержимое папки необходимо перенести в директорию dirprm для дальнейшей настройки Replicat.

Из скопированных файлов открываем custom_kafka_producer.properties, здесь нужно указать адрес сервера Kafka:

bootstrap.servers=kafka:9092 acks=1 reconnect.backoff.ms=1000 value.serializer=org.apache.kafka.common.serialization.ByteArraySerializer key.serializer=org.apache.kafka.common.serialization.ByteArraySerializer # 100KB per partition batch.size=16384 linger.ms=0

Следующим шагом открываем файл kafka.props и в параметре gg.classpath прописываем путь до библиотек Kafka. Также здесь можно указать формат выгружаемых данных:  xml, delimitedtext, json, json_row, avro_row, avro_op. Формат данных определяется разработчиками или техзаданием.

gg.handlerlist = kafkahandler gg.handler.kafkahandler.type=kafka gg.handler.kafkahandler.KafkaProducerConfigFile=custom_kafka_producer.properties #The following resolves the topic name using the short table name gg.handler.kafkahandler.topicMappingTemplate=${tableName} #The following selects the message key using the concatenated primary keys gg.handler.kafkahandler.keyMappingTemplate=${primaryKeys} gg.handler.kafkahandler.format=avro_op gg.handler.kafkahandler.SchemaTopicName=mySchemaTopic gg.handler.kafkahandler.BlockingSend =false gg.handler.kafkahandler.includeTokens=false gg.handler.kafkahandler.mode=op gg.handler.kafkahandler.MetaHeaderTemplate=${alltokens}     goldengate.userexit.writers=javawriter javawriter.stats.display=TRUE javawriter.stats.full=TRUE   gg.log=log4j gg.log.level=INFO   gg.report.time=30sec   #Sample gg.classpath for Apache Kafka gg.classpath=dirprm/:/u01/kafka_2.13-2.7.0/libs/* #Sample gg.classpath for HDP #gg.classpath=/etc/kafka/conf:/usr/hdp/current/kafka-broker/libs/*

В файле rkafka.prm указываем нужные таблицы для репликации:

REPLICAT rkafka -- Trail file for this example is located in "AdapterExamples/trail" directory -- Command to add REPLICAT -- add replicat rkafka, exttrail AdapterExamples/trail/tr TARGETDB LIBFILE libggjava.so SET property=dirprm/kafka.props REPORTCOUNT EVERY 1 MINUTES, RATE GROUPTRANSOPS 10000 MAP appuser.test01, TARGET appuser.test01;

После настроек остается только добавить Replicat привычным способом:

GGSCI> add replicat rkafka, exttrail /u01/app/oracle/product/19.1.0/oggcore_1/dirdat/ex start rkafka

После включения данные начнут передаваться — это можно увидеть со стороны Kafka:

bash> bin/kafka-topics.sh --list --bootstrap-server kafka:2181 TEST01 __consumer_offsets mySchemaTopic quickstart-events bash> bin/kafka-console-consumer.sh --topic TEST01 --from-beginning --bootstrap-server kafka:9092 APPUSER.TEST01I42021-01-22 11:32:20.00000042021-01-22T15:20:51.081000(00000000000000001729ID$@Ten APPUSER.TEST01I42021-01-22 15:24:55.00000042021-01-22T15:25:00.665000(00000000000000001872ID$@One

Репликация данных Oracle→Oracle с помощью Oracle GoldenGate

Примечание автора: нам важно сконфигурировать репликацию как downstream, то есть как вынесенную систему, для отслеживания изменений. Делается это в первую очередь для увеличения отказоустойчивости системы и снижения нагрузки на сервер источника. Обязательное требование при этом — одинаковая архитектура серверов. То есть невозможно будет настроить прием логов на downstream mining БД, которая расположена на Solaris Sparc64, при том что БД-источник на Linux x86-64. Если планируется использовать REAL_TIME_CAPTURE (репликация в реальном времени), то на Mining БД нужно создать standby-redo-логи такого же размера, как и на БД-источнике. Число логов на один больше, чем на источнике. 

Включаем режим archivelog, forcelogging, supplemental_logging на БД-источнике:

SQL ALTER DATABASE ARCHIVELOG; ALTER DATABASE FORCE LOGGING; ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;

К сведению инженера: для включения этого режима лучше перезагрузить БД и не выполнять на бою. Включение может привести к деградации производительности вплоть до остановки работы СУБД. Активация режима FORCE LOGGING принудительно логирует все изменения, даже те, которые выполнялись с опцией NOLOGGING. Режим Supplemental Log data добавляет метаданные в вектора изменений, которые позволяют работать режиму CDC. Включение этих режимов может привести к увеличению генерируемых изменений БД — генерации архивных логов. По нашему опыту, рост составляет 15–25%.

Проверяем:

SQL> SELECT log_mode, supplemental_log_data_min, force_logging FROM v$database;

Создаем пользователя для репликации на источнике и downstream БД:

SQL> create user ggadmшn identified by ggadmin default tablespace users temporary tablespace temp;exec dbms_goldengate_auth.grant_admin_privilege (' ggadmшn'); alter system set enable_goldengate_replication=TRUE scope=both;

Добавляем db_unique_name downstream БД в log_archive_config на источнике:

SQL> ALTER SYSTEM SET LOG_ARCHIVE_CONFIG='DG_CONFIG=([SOURCEDB], [MININGDB])' scope=both;

Копируем файл паролей с источника на Mining DB. Здесь добавляем LAD для хранения архивных логов источника и собственных архивных логов:

SQL> ALTER SYSTEM SET LOG_ARCHIVE_DEST_1='LOCATION=[archive_log_location] VALID_FOR=(ONLINE_LOGFILE,PRIMARY_ROLE)' scope=both; ALTER SYSTEM SET LOG_ARCHIVE_DEST_2='LOCATION=[archive_log_location]/REMOTE VALID_FOR=(STANDBY_LOGFILE,PRIMARY_ROLE)' scope=both; ALTER SYSTEM SET LOG_ARCHIVE_DEST_state_2=enable scope=both;

На источнике и Mining DB добавляем TNS записи БД. Проверяем, что сетевой доступ открыт и удается подключиться с каждого сервера с БД на каждый.

Добавляем новый LAD для downstream БД:

SQL> ALTER SYSTEM SET LOG_ARCHIVE_DEST_2='SERVICE=[MININGDB_SERVICE] ASYNC NOREGISTER VALID_FOR=(ONLINE_LOGFILES,PRIMARY_ROLE) DB_UNIQUE_NAME=[MININGDB]' scope=both; ALTER SYSTEM SET LOG_ARCHIVE_DEST_STATE_2=enable scope=both;

Проверяем, что нет ошибок передачи логов:

select destination,status,error from v$archive_dest_status where rownum < 3;

Устанавливаем Oracle GoldenGate нужной версии на сервер с Mining DB и выполняем первоначальную настройку. Добавляем credential для подключения к источнику:

GGSCI> Create subdirs – создаем необходимую структуру директорий edit param MGR – правим параметры менеджера     PORT 7800 GGSCI> start MGR – запускаем менеджер create wallet – создаем валлет, где будут хранится параметры аутентификации пользователей  Created wallet at location 'dirwlt'. Opened wallet at location 'dirwlt'.GGSCI> add credentialstore Credential store created in ./dircrd/.  GGSCI> alter credentialstore add user ggadmin@[sourcedb_tns] alias ggadmin_[sourcedb] Password: *********** Credential store in ./dircrd/ altered.  GGSCI> alter credentialstore add user ggadmin@[mining_tns] alias ggadmin_[miningdb] Password: *********** Credential store in ./dircrd/ altered.  GGSCI> dblogin useridalias ggadmin_[sourcedb] – пробуем залогинится в БД источник Successfully logged into database.

На БД-источнике включаем режим расширенного логирования для таблиц, которые будем реплицировать:

SQL> alter table [OWNER].[TABLE_NAME] add supplemental log data (all) columns;

Данная команда требует эксклюзивной блокировки на время выполнения. Рекомендуем выполнять в период наименьшей нагрузки.

Настраиваем процессы Extract и PUMP для получения и передачи изменений:

GGSCI> dblogin useridalias ggadmin_[sourcedb] Successfully logged into database. GGSCI> miningdblogin useridalias ggadmin_[miningdb] Successfully logged into mining database.  GGSCI> add extract eSOURCEDB integrated tranlog begin now – добавляем процесс экстракт (процесс, собирающий изменения) EXTRACT added. GGSCI> add exttrail ./dirdat/SOURCEDB/lt extract eSOURCEDB – добавляем файловый путь, куда будут складываться изменения, добытые нашим экстрактом EXTTRAIL added. GGSCI> add extract pSOURCEDB exttrailsource ./dirdat/SOURCEDB/lt – создаем необязательный процесс PUMP, который будет передавать добытые изменения на другой сервер.  EXTRACT added. GGSCI> add rmttrail ./dirdat/SOURCEDB/rt extract pSOURCEDB RMTTRAIL added. GGSCI> register extract eSOURCEDB database Extract ESOURCEDB successfully registered with database at SCN 261106958. GGSCI> add replicat rSOURCEDB integrated exttrail ./dirdat/SOURCEDB/rt REPLICAT (Integrated) added.  GGSCI> edit param eSOURCEDB     EXTRACT eSOURCEDB     useridalias ggadmin_[sourcedb]     TRANLOGOPTIONS MININGUSERALIAS ggadmin_[miningdb]     TRANLOGOPTIONS INTEGRATEDPARAMS (downstream_real_time_mine Y)     EXTTRAIL ./dirdat/SOURCEDB/lt     TABLE [OWNER].[TABLE_NAME]; GGSCI> edit param pSOURCEDB     EXTRACT pSOURCEDB     useridalias ggadmin_[sourcedb]     RMTHOST host02, MGRPORT 7800     RMTTRAIL ./dirdat/SOURCEDB/rt     PASSTHRU     TABLE *.*;

Запускаем процесс Extract и PUMP:

GGSCI> start  eSOURCEDB  start  pSOURCEDB 

Перейдем к приемнику.

На БД-приемнике устанавливаем и конфигурируем Oracle GoldenGate и создаем процесс репликата:

GGSCI> edit param rSOURCEDB      REPLICAT rSOURCEDB     ASSUMETARGETDEFS     USERIDALIAS oggsuser_PRODDB     MAP [OWNER].[TABLE_NAME, TARGET [TARGET_OWNER].[TARGET_TABLE_NAME], FILTER (@getenv ('TRANSACTION' , 'CSN') > SCN_консистентности)

Выполняем первичную инициализацию реплицируемой таблицы.
Для уверенности, что все изменения будут перенесены при первичной инициализации, мы временно блокируем таблицу, получаем SCN, когда точно никакие процессы не меняют данные (т.н. SCN-консистентности). С него потом запускаем репликацию.

Получаем SCN-консистентности для таблицы на источнике:

SQL> SET SERVEROUTPUT ON; declare v_tab_name varchar2(32000); v_scn number; v_ku$status ku$_status1020; begin execute immediate 'alter session set ddl_lock_timeout=20'; for i in (select * from dba_tables dt where dt.owner = '[OWNER]' and dt.table_name in ('[TABLE_NAME]'} loop execute immediate 'lock table '||i.owner||'.'||i.table_name||' in share mode'; v_tab_name := v_tab_name || ','''||i.table_name||''''; end loop; v_tab_name := ltrim(v_tab_name,','); select current_scn into v_scn from v$database; dbms_lock.sleep(3); commit; dbms_output.put_line('SCN: '||to_char(v_scn)); dbms_output.put_line('FILTER (@getenv (''TRANSACTION'' , ''CSN'') > '||to_char(v_scn)||')'); end;

Выгружаем таблицу с источника при помощи datapump. Примерный файл параметров ниже:

USERID="/ as sysdba" DIRECTORY=EXPGG LOGFILE=ibs_zrecords.log DUMPFILE=ibs_zrecords%U.dmp TABLES=IBS.Z\#RECORDS exclude=RLS_POLICY,CONSTRAINT,GRANT,REF_CONSTRAINT,STATISTICS,TRIGGER,INDEX FLASHBACK_SCN={SCN} STATUS=60 PARALLEL=4

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

Создаем Primary KEY и другие индексы, если нужно.

Заметка авторов на полях: уникальные ключи PK или UK являются необходимыми для работы репликации.

Запускаем репликат на приемнике:

GGSCI> start  rSOURCEDB  info rSOURCEDB 

Контролируем работу процессов и устраняем возникающие ошибки. Готово.

Репликация данных Sybase→Oracle с помощью Sybase RS

Для создания нового коннекта и репликации таблиц из Sybase в Oracle выполняем следующие шаги.

Создаем на стороне Оracle приемную схему:

SQL>

create user USERNAME identified by PASSWORD default tablespace USERS temporary tablespace TEMP quota unlimited on USERS;

Затем заводим в базе Oracle пользователя, под кем реплика будет подключаться к БД:

SQL>

create user USERNAME_RS identified by PASSWORD default tablespace USERS temporary tablespace TEMP quota unlimited on USERS;

Создаем в Oracle в нужной схеме приемные таблицы нужной структуры.
Убеждаемся, что на стороне Оracle в таблицах создан PK или уникальный индекс. Если нет — создаем.
Убеждаемся в наличии сетевого доступа с RS на новый Oracle БД.
Прописываем на RS в файле interfaces новое подключение к приемной базе Оracle с помощью уже настроенного отдельного ECDA сервера. 

Пример записи в файле interfaces:

dco_tst_01         query tcp ether server-name 9035

Данная запись означает, что на ECDA сервере server-name, на порту 9035, настроено подключение к нужной приемной базе, и данное подключение называется dco_tst_01

ECDA (Enterprise Connect Data Access) — это гетерогенный шлюз (gateway), который является отдельно приобретаемым и отдельно лицензируемым продуктом для репликации Sybase баз данных в Non-ASE базы данных. ECDA позволяет установить драйвера для подключения к различным СУБД, таким как Oracle, MSSQL, PostgreSQL и др., и реплицировать в другие СУБД данные из Sybase ASE. Драйверы для каждой новой СУБД приобретаются отдельно. Инсталляция и настройка ECDA — отдельная тема и в данной статье рассматриваться не будет.

В Sybase RS 15.5 и позже появилась возможность напрямую подключаться через ECO (ExpressConnect for Oracle) в Oracle БД без разворачивания ECDA. Для этого необходимо прописать локально на RS параметры подключения к Oracle в файле:

$RS_HOME/REP-16_0/connector/oraoci/network/admin/tnsnames.ora SOURCEDB=   (DESCRIPTION =     (ADDRESS_LIST =       (ADDRESS = (PROTOCOL = TCP)(HOST = hostname)(PORT = 1521))     )     (CONNECT_DATA =       (SERVER = DEDICATED)       (SERVICE_NAME = SOURCEDB)     )   )

 Настраиваем репликации. Создаем новый connection в RS для указания приемника репликационных данных (сервер и база).

-- Create Database Connection eco.dwh create connection to "eco"."dwh"     set error class to "err_class_IgnorDubli"     set function string class to "rs_oracle_function_class"     set username "USERNAME_RS"     set password "PASSWORD_USERNAME_RS"  go -- Set connection configuration parameters for eco.dwh alter connection to eco.dwh set cmd_direct_replicate to 'off' alter connection to eco.dwh set exec_sqm_write_request_limit to '4194304' 

Создаем replication definition в RS. Немного о терминах: replication definition — это специальный объект на RS-сервере, который описывает, из какой схемы, какая таблица –> в какую приемную схему и таблицу будет реплицироваться. Описываются, какие поля в таблице, типы данных и размерность подлежат реплицированию. Также в replication definition указан PK, который будет использоваться системой репликации для идентификации строк. Одним словом, это «правила» для репликационного сервера, какие таблицы опубликованы для репликации и как будут передаваться данные.

Первой создаем KEY repdef: 

create replication definition RDNAME dFRONT_KEY_f_oper_card  with primary at SOURCE  with primary table named TABLE_NAME  with replicate table named TARGET_TABLE_NAME  ("f_oper" int)  primary key("f_oper")  replicate minimal columns go

Затем — основной repdef новой таблицы:

create replication definition  RDNAME  with primary at SOURCEDB  with primary table named TABLE_NAME  with replicate table named TARGET_TABLE_NAME(     "f_oper"           int,               "f_req"            int,               "f_cardhandreason" smallint,          "type_return"      tinyint,          "inn"              varchar(12),       "acc_num"          varchar(20),       "bik"              varchar(10),       "service_name"     varchar(100),      "service_adress"   varchar(100),      "date_decision"    date,              "way_decision"     varchar(255),      "contact_decision" varchar(255),      "result_decision"  varchar(255),      "comment_reason"   varchar(255),      "descr"            varchar(100)  )  primary key("f_oper")  replicate minimal columns go 

Важно. Если в таблице ключевое поле IDENTITY, создаем KEY repdef только для одного поля с типом IDENTITY. 

create replication definition RDNAME     with primary at SOURCEDB     with primary table named TABLE_NAME      with replicate table named TARGET_TABLE_NAME     (f_kurs identity)     primary key (f_kurs)     replicate minimal columns     with dynamic sql

Основной repdef должен быть «identity map to rs_oracle_float» (или «ident_id identity map to numeric») — в Oracle нет типа IDENTITY. Здесь это поле должно быть типом NUMBER(10):

create replication definition RDNAME     with primary at SOURCEDB     with primary table named TABLE_NAME      with replicate table named TARGET_TABLE_NAME     (f_kurs identity map to rs_oracle_float,          c_kurstype tinyint,          c_currency_1 smallint,          c_currency_2 smallint,          f_bank int,          f_user int,          rate numeric,          date_begin datetime,          date_create datetime,          username varchar(30),          workstation varchar(50),          f_office int,          special tinyint) primary key (f_kurs) replicate minimal columns with dynamic sql

Создаем подписку на созданный основной repdef на RS:

create subscription SUBS_NAME  for RDNAME  with replicate at eco.dwh without materialization go

Теперь включаем репликацию новой таблицы в Sybase Front. После выполнения нижеуказанной команды RepAgent начнет отслеживать в транзакционном логе Sybase базы источника все изменения по указанной таблице и передавать их на RS для дальнейшей обработки и передачи их на приемник:

sp_setreptable SOURCE_TABLE_NAME, true go

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

rem %1-Server rem %2-Passw_SA rem %3-Database Name rem %4-Table Name  bcp.exe %3.dbo.%4 out %4.dat -c -td6vWf7 -rd1K6t7 -Jcp1251 -eErr.txt -S%1 -Usa -P%2

Создаем файл загрузки данных в Oracle и загружаем данные через sqlldr.

Пример загрузки таблицы F_KURS.ctl:

LOAD DATA CHARACTERSET CL8MSWIN1251 INFILE 'f_kurs.dat' "STR 'd1K6t7'" BADFILE 'f_kurs.BAD' DISCARDFILE 'f_kurs.DSC' APPEND INTO TABLE TARGET_TABLE_NAME FIELDS TERMINATED BY "d6vWf7" TRAILING NULLCOLS ( F_KURS           CHAR(255), C_KURSTYPE       CHAR(255), C_CURRENCY_1     CHAR(255), C_CURRENCY_2     CHAR(255), F_BANK           CHAR(255), F_USER           CHAR(255), RATE             CHAR(255), DATE_BEGIN       CHAR(255) "TO_DATE(SUBSTR(:DATE_BEGIN,1,20)||SUBSTR(:DATE_BEGIN,25,2),'Mon DD YYYY HH:MI:SSAM')", DATE_CREATE      CHAR(255) "TO_DATE(SUBSTR(:DATE_CREATE,1,20)||SUBSTR(:DATE_CREATE,25,2),'Mon DD YYYY HH:MI:SSAM')", USERNAME         CHAR(255), WORKSTATION      CHAR(255), F_OFFICE         CHAR(255), SPECIAL          CHAR(255) )   bash#: sqlldr USERNAME/PASSW control={location}/F_KURS.ctl 

Поднимаем очередь, смотрим и устраняем возникающие ошибки репликации:

resume connection to server.db

На этом все! Делитесь в комментариях, что получилось, а что — нет. Какой репликацией чаще всего пользуетесь?

До встречи!