¿Cómo puedo replicar tablas en RDS para PostgreSQL mediante replicación lógica?

Resolución

Un caso de uso típico de replicación lógica es replicar un conjunto de tablas entre dos instancias de base de datos de Amazon RDS para PostgreSQL. RDS para PostgreSQL admite la replicación lógica con PostgreSQL 10.4 y versiones posteriores. La versión 2.2.0 y posteriores de la edición compatible con PostgreSQL de Amazon Aurora admite la replicación lógica con la versión 10.6 y posteriores de PostgreSQL.

En la resolución proporcionada, dos tablas de origen se replican en dos tablas de destino mediante la replicación lógica en RDS para PostgreSQL. La replicación lógica realiza primero la carga inicial de los datos que ya están presentes en las tablas de origen y, a continuación, continúa replicando los cambios en curso.

Activar la replicación lógica

Para activar la replicación lógica en RDS para PostgreSQL, modifique un grupo de parámetros personalizado para establecer rds.logical_replication en 1 y conecte rds.logical_replication a la instancia de base de datos. Actualice el grupo de parámetros para establecer rds.logical_replication en 1 si hay un grupo de parámetros personalizado conectado a una instancia de base de datos. El parámetro rds.logical_replication es un parámetro estático que requiere el reinicio de la instancia de base de datos para que surta efecto. Cuando se reinicia la instancia de base de datos, el parámetro wal_level se establece en logical.

Verifique los valores de wal_level y 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)

Conectarse a la base de datos de origen en la instancia de base de datos de origen

Conéctese a la base de datos de origen en la instancia de base de datos RDS para PostgreSQL de origen. Cree las tablas de origen:

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

Inserte datos en las tablas de origen:

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

Crear una publicación para las tablas de origen

Cree una publicación para las tablas de origen. Utilice una consulta SELECT para comprobar los detalles de la publicación que se creó:

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)

Compruebe que las tablas de origen se hayan agregado a la publicación:

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

Nota: Para replicar todas las tablas de una base de datos, ejecute este comando:

CREATE PUBLICATION testpub FOR ALL TABLES;

Conectarse a la base de datos de destino y crear tablas de destino

Conéctese a la base de datos de destino en la instancia de base de datos objetivo. Cree las tablas de destino con los mismos nombres que las tablas de origen. Asegúrese de que no haya datos presentes en las tablas de destino mediante la ejecución de una consulta SELECT en las tablas 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)

Crear y verificar la suscripción en la base de datos objetivo

Cree la suscripción en la base de datos de destino. Utilice una consulta SELECT para comprobar si la suscripción 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 almacenar una versión en texto sin formato de su nombre de usuario y contraseña en los registros de la base de datos, antes de crear la suscripción, ejecute los siguientes comandos:

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

Cuando se crea la suscripción, la suscripción carga todos los datos presentes en las tablas de origen en las tablas de destino. Ejecute una consulta SELECT en las tablas de destino para comprobar que se cargan los datos iniciales:

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

Verificar la ranura de replicación en la base de datos de origen

La creación de una suscripción en la base de datos de destino crea una ranura de replicación en la base de datos de origen. Verifique los detalles de la ranura de replicación mediante la ejecución de la siguiente consulta SELECT en la base de datos de origen:

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)

Probar la replicación desde tablas de origen

Compruebe si los cambios de datos en las tablas de origen se replican en las tablas de destino mediante la inserción de filas en las tablas de origen:

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)

Probar la replicación mediante la verificación del número de filas en las tablas de destino

Verifique el número de filas en las tablas de destino para confirmar que se están replicando nuevas inserciones en las tablas de destino:

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

Limpiar y desactivar la replicación lógica

Limpie y desactive la replicación lógica cuando la replicación esté completa y ya no sea necesaria. Las ranuras de replicación inactivas provocan la acumulación de archivos WAL en las instancias de base de datos de origen. Los archivos WAL pueden ocupar espacio de almacenamiento y provocar interrupciones.

Eliminar la suscripción en la base de datos 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)

Nota: Al cancelar la suscripción, también se pierde la ranura de replicación que creó la suscripción.

Verifique que la ranura de replicación se haya eliminado de la base de datos de origen mediante la ejecución de la siguiente sentencia de consulta SELECT en el origen:

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)

Elimine la publicación. Compruebe que la publicación se ha elimine correctamente:

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 el valor de rds.logical_replication a 0 en el grupo de parámetros personalizados conectado a la instancia de base de datos. Reinicie la instancia de base de datos según sea necesario si la instancia de base de datos no utiliza la replicación lógica.

Revise max_replication_slots, max_wal_senders, max_logical_replication_workers, max_worker_processes, and max_sync_workers_per_subscription en función de su uso.

Nota: Los siguientes comandos comprueban si hay ranuras de replicación inactivas, determinan los tamaños respectivos de las ranuras y, a continuación, descartan las ranuras, si es necesario.

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');

---

Información relacionada

Documentación de PostgreSQL para Replication (Replicación)