Como faço para replicar tabelas no RDS para PostgreSQL usando a replicação lógica?

Resolução

Um caso de uso típico para replicação lógica é replicar um conjunto de tabelas entre duas instâncias de banco de dados do Amazon RDS para PostgreSQL. O RDS para PostgreSQL oferece suporte à replicação lógica com o PostgreSQL 10.4 e posterior. O Amazon Aurora PostgreSQL Compatible Edition versão 2.2.0 e posterior oferece suporte à replicação lógica com o PostgreSQL 10.6 e posterior.

Na resolução fornecida, duas tabelas de origem são replicadas para duas tabelas de destino usando replicação lógica no RDS para PostgreSQL. A replicação lógica primeiro executa a carga inicial de dados já presente nas tabelas de origem e, em seguida, continua replicando as alterações contínuas.

Ativar a replicação lógica

Para ativar a replicação lógica no RDS para PostgreSQL, modifique um grupo de parâmetro personalizado para definir rds.logical_replication como 1 e anexe rds.logical_replication à instância de banco de dados. Atualize o grupo de parâmetro para definir rds.logical_replication como 1 se um grupo de parâmetro personalizado estiver anexado a uma instância de banco de dados. O parâmetro rds.logical_replication é um parâmetro estático que requer uma reinicialização da instância de banco de dados para entrar em vigor. Quando a instância de banco de dados é reinicializada, o parâmetro wal_level é definido como logical.

Verifique os valores para wal_level e rds.logical_replication:

postgres=> SELECT name,setting FROM pg_settings WHERE name IN ('wal_level','rds.logical_replication');
          name           | setting
-------------------------+---------
 rds.logical_replication | on
 wal_level               | logical
(2 rows)

Conecte-se ao banco de dados de origem na instância de banco de dados

Conecte-se ao banco de dados de origem na instância de banco de dados RDS para PostgreSQL de origem. Crie as tabelas de origem:

source=> CREATE TABLE reptab1 (slno int primary key);
CREATE TABLE
source=> CREATE TABLE reptab2 (name varchar(20));
CREATE TABLE

Insira dados nas tabelas de origem:

source=> INSERT INTO reptab1 VALUES (generate_series(1,1000));
INSERT 0 1000
source=> INSERT INTO reptab2 SELECT SUBSTR ('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789',((random()*(36-1)+1)::integer),1) FROM generate_series(1,50);
INSERT 0 50

Criar uma publicação para tabelas de origem

Crie uma publicação para as tabelas de origem. Use uma consulta SELECT para verificar os detalhes da publicação que foi criada:

source=> CREATE PUBLICATION testpub FOR TABLE reptab1,reptab2;
CREATE PUBLICATION
source=> SELECT * FROM pg_publication;
  oid   | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot
--------+---------+----------+--------------+-----------+-----------+-----------+-------------+------------
 115069 | testpub |    16395 | f            | t         | t         | t         | t           | f
(1 row)

Verifique se as tabelas de origem foram adicionadas à publicação:

source=> SELECT * FROM pg_publication_tables;
 pubname | schemaname | tablename
---------+------------+-----------
 testpub | public     | reptab1
 testpub | public     | reptab2
(2 rows)

Observação: Para replicar todas as tabelas em um banco de dados, execute este comando:

CREATE PUBLICATION testpub FOR ALL TABLES;

Conectar-se ao banco de dados de destino e criar tabelas de destino

Conecte-se ao banco de dados de destino na instância de banco de dados de destino. Crie as tabelas de destino com os mesmos nomes das tabelas de origem. Certifique-se de que não haja dados presentes nas tabelas de destino executando uma consulta SELECT nas tabelas de destino:

target=> CREATE TABLE reptab1 (slno int primary key);
CREATE TABLE
target=> CREATE TABLE reptab2 (name varchar(20));
CREATE TABLE
target=> SELECT count(*) FROM reptab1;
 count
-------
     0
(1 row)
target=> SELECT count(*) FROM reptab2;
 count
-------
     0
(1 row)

Criar e verificar a assinatura no banco de dados de destino

Crie a assinatura no banco de dados de destino. Use uma consulta SELECT para verificar se a assinatura está habilitada:

target=> CREATE SUBSCRIPTION testsub CONNECTION 'host=<source RDS/host endpoint> port=5432 dbname=<source_db_name> user=<user> password=<password>' PUBLICATION testpub;
NOTICE:  Created replication slot "testsub" on publisher
CREATE SUBSCRIPTION
target=> SELECT oid,subname,subenabled,subslotname,subpublications FROM pg_subscription;
  oid  | subname | subenabled | subslotname | subpublications
-------+---------+------------+-------------+-----------------
 16434 | testsub | t          | testsub     | {testpub}
(1 row)

Importante: Para evitar armazenar uma versão em texto simples do seu nome de usuário e senha nos logs do banco de dados, antes de criar a assinatura, execute os seguintes comandos:

target=> SET log_min_messages to 'PANIC';
SET
target=> SET log_statement to NONE;
SET

Quando a assinatura é criada, ela carrega todos os dados presentes nas tabelas de origem para as tabelas de destino. Execute uma consulta SELECT nas tabelas de destino para verificar se os dados iniciais são carregados:

target=> SELECT count(*) FROM reptab1;
 count
-------
  1000
(1 row)
target=> SELECT count(*) FROM reptab2;
 count
-------
    50
(1 row)

Verificar o slot de replicação no banco de dados de origem

A criação de uma assinatura no banco de dados de destino cria um slot de replicação no banco de dados de origem. Verifique os detalhes do slot de replicação executando a seguinte consulta SELECT no banco de dados de origem:

source=> SELECT * FROM pg_replication_slots;
 slot_name |  plugin  | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size
 ----------+----------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+---------------
 testsub   | pgoutput | logical   | 115048 | source   | f         | t      |        846 |      |         6945 | 58/B4000568 | 58/B40005A0         | reserved   |
(1 row)

Testar a replicação das tabelas de origem

Teste se as alterações de dados nas tabelas de origem estão sendo replicadas para as tabelas de destino inserindo linhas nas tabelas de origem:

source=> INSERT INTO reptab1 VALUES(generate_series(1001,2000));
INSERT 0 1000
source=> INSERT INTO reptab2 SELECT SUBSTR('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789',((random()*(36-1)+1)::integer),1) FROM generate_series(1,50);
INSERT 0 50
source=> SELECT count(*) FROM reptab1;
 count
-------
  2000
(1 row)
source=> SELECT count(*) FROM reptab2; count
-------
   100
(1 row)

Testar a replicação verificando o número de linhas nas tabelas de destino

Verifique o número de linhas nas tabelas de destino para confirmar se novas inserções estão sendo replicadas para as tabelas de destino:

target=> SELECT count(*) FROM reptab1;
 count
-------
  2000
(1 row)
target=> SELECT count(*) FROM reptab2;
 count
-------
   100
(1 row)

Limpar e desativar a replicação lógica

Limpe e desative a replicação lógica quando a replicação estiver concluída e não for mais necessária. Os slots de replicação inativos causam o acúmulo de arquivos WAL nas instâncias de banco de dados de origem. Os arquivos WAL podem encher o armazenamento e causar interrupções.

Descarte a assinatura no banco de dados de destino:

target=> DROP SUBSCRIPTION testsub;
NOTICE:  Dropped replication slot "testsub" on publisher
DROP SUBSCRIPTION
target=> SELECT * FROM pg_subscription;
oid | subdbid | subname | subowner | subenabled | subconninfo | subslotname | subsynccommit | subpublications
----+---------+---------+----------+------------+-------------+-------------+---------------+-----------------
(0 rows)

Observação: A eliminação da assinatura também descarta o slot de replicação que a assinatura criou.

Verifique se o slot de replicação foi eliminado do banco de dados de origem executando a seguinte instrução de consulta SELECT na origem:

source=> SELECT * FROM pg_replication_slots;
slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size
----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+--------------
(0 rows)

Abandone a publicação. Verifique se a publicação foi descartada:

source=> DROP PUBLICATION testpub;
DROP PUBLICATION
source=> SELECT * FROM pg_publication;
 oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot
-----+---------+----------+--------------+-----------+-----------+-----------+-------------+------------
(0 rows)
source=> SELECT * FROM pg_publication_tables;
 pubname | schemaname | tablename
---------+------------+-----------
(0 rows)

Modifique rds.logical_replication para 0 no grupo de parâmetro personalizado que está anexado à instância de banco de dados. Reinicialize a instância de banco de dados conforme necessário se a instância de banco de dados não estiver usando replicação lógica.

Analise max_replication_slots, max_wal_senders, max_logical_replication_workers, max_worker_processes e max_sync_workers_per_subscription com base no seu uso.

Observação: Os comandos a seguir verificam se há slots de replicação inativos, determinam os tamanhos dos slots e, em seguida, eliminam os slots, se necessário.

SELECT slot_name, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(),restart_lsn)) AS replicationSlotLag, active FROM pg_replication_slots ;

SELECT pg_drop_replication_slot('Your_slotname_name');

---

Informações relacionadas

Documentação do PostgreSQL para Replicação