Wie repliziere ich Tabellen in RDS für PostgreSQL mithilfe der logischen Replikation?

Lösung

Ein typischer Anwendungsfall fürlogische Replikation ist das Replizieren einer Reihe von Tabellen zwischen zwei Amazon RDS für PostgreSQL-DB-Instances. RDS for PostgreSQL unterstützt die logische Replikation mit PostgreSQL 10.4 und höher. Amazon Aurora PostgreSQL-Compatible Edition Version 2.2.0 und höher unterstützt die logische Replikation mit PostgreSQL 10.6 und höher.

In der bereitgestellten Auflösung werden zwei Quelltabellen mithilfe der logischen Replikation in RDS für PostgreSQL in zwei Zieltabellen repliziert. Bei der logischen Replikation werden zunächst die bereits in den Quelltabellen vorhandenen Daten initial geladen und dann die laufenden Änderungen repliziert.

Logische Replikation einschalten

Um die logische Replikation in RDS für PostgreSQL zu aktivieren, ändern Sie eine benutzerdefinierte Parametergruppe, umrds.logical_replication auf 1 zu setzen, und fügen Sierds.logical_replication an die DB-Instance an. Aktualisieren Sie die Parametergruppe, umrds.logical_replication auf 1 zu setzen, wenn eine benutzerdefinierte Parametergruppe an eine DB-Instance angehängt ist. Der rds.logical_replication Parameter ist ein statischer Parameter, der einen Neustart der DB-Instance erfordert, damit er wirksam wird. Wenn die DB-Instance neu gestartet wird, wird der Parameter wal_level auflogical gesetzt.

Überprüfen Sie die Werte fürwal_level undrds.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)

Verbindung zur Quelldatenbank in der Quell-DB-Instance herstellen

Stellen Sie eine Verbindung mit der Quelldatenbank in der Quell-RDS für PostgreSQL-DB-Instance her. Erstellen Sie die Quelltabellen:

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

Fügen Sie Daten in die Quelltabellen ein:

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

Erstellen einer Publikation für Quelltabellen

Erstellen Sie eine Publikation für die Quelltabellen. Verwenden Sie eine SELECT-Abfrage, um die Details der erstellten Publikation zu überprüfen:

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)

Stellen Sie sicher, dass die Quelltabellen der Publikation hinzugefügt wurden:

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

Hinweis: Um alle Tabellen in einer Datenbank zu replizieren, führen Sie diesen Befehl aus:

CREATE PUBLICATION testpub FOR ALL TABLES;

Verbindung zur Zieldatenbank herstellen und Zieltabellen erstellen

Stellen Sie eine Verbindung zur Zieldatenbank in der Ziel-DB-Instance her. Erstellen Sie die Zieltabellen mit den gleichen Namen wie die Quelltabellen. Stellen Sie sicher, dass in den Zieltabellen keine Daten vorhanden sind, indem Sie eine SELECT-Abfrage für die Zieltabellen ausführen:

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)

Abonnement in der Zieldatenbank erstellen und überprüfen

Erstellen Sie das Abonnement in der Zieldatenbank. Überprüfen Sie mit einer SELECT-Abfrage, ob das Abonnement aktiviert ist:

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)

Wichtig: Um zu vermeiden, dass eine Klartext-Version Ihres Benutzernamens und Kennworts in den Datenbankprotokollen gespeichert wird, führen Sie vor dem Erstellen des Abonnements die folgenden Befehle aus:

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

Wenn das Abonnement erstellt wird, lädt das Abonnement alle in den Quelltabellen vorhandenen Daten in die Zieltabellen. Führen Sie eine SELECT-Abfrage für die Zieltabellen aus, um sicherzustellen, dass die ursprünglichen Daten geladen werden

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

Überprüfung des Replikations-Slots in der Quelldatenbank

Die Erstellung eines Abonnements in der Zieldatenbank erzeugt einen Replikationsslot in der Quelldatenbank. Überprüfen Sie die Details des Replikationsslots, indem Sie die folgende SELECT-Abfrage in der Quelldatenbank ausführen:

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)

Testen der Replikation aus Quelltabellen

Testen Sie, ob Datenänderungen in den Quelltabellen in die Zieltabellen repliziert werden, indem Sie Zeilen in die Quelltabellen einfügen:

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)

Testen Sie die Replikation, indem Sie die Anzahl der Zeilen in Zieltabellen überprüfen

Überprüfen Sie die Anzahl der Zeilen in den Zieltabellen, um sicherzustellen, dass neue Einfügungen in die Zieltabellen repliziert werden:

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

Bereinigen und Ausschalten der logischen Replikation

Bereinigen und deaktivieren Sie die logische Replikation, wenn die Replikation abgeschlossen ist und nicht mehr benötigt wird. Inaktive Replikationsslots verursachen die Ansammlung von WAL-Dateien auf den Quell-DB-Instanzen. WAL-Dateien können Speicherplatz füllen und Ausfälle verursachen.

Löschen Sie das Abonnement in der Zieldatenbank:

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)

Hinweis: Durch das Löschen des Abonnements wird auch der vom Abonnement erstellte Replikationsslot gelöscht.

Überprüfen Sie, ob der Replikationsslot aus der Quelldatenbank gelöscht wurde, indem Sie die folgende SELECT-Abfrageanweisung in der Quelle ausführen:

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)

Löschen Sie die Publikation. Stellen Sie sicher, dass die Veröffentlichung erfolgreich gelöscht wurde:

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)

Ändern Sierds.logical_replication in der benutzerdefinierten Parametergruppe, die an die DB-Instance angehängt ist, auf0. Starten Sie die DB-Instance nach Bedarf neu, wenn die DB-Instance keine logische Replikation verwendet.

Überprüfen Siemax_replication_slots, max_wal_senders, max_logical_replication_workers, max_worker_processes, und max_sync_workers_per_subscription basierend auf Ihrer Nutzung.

Hinweis: Mit den folgenden Befehlen wird geprüft, ob inaktive Replikationsslots vorhanden sind, die jeweilige Größe der Slots bestimmt und die Slots dann gegebenenfalls gelöscht.

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

---

Relevante Informationen

PostgreSQL-Dokumentation für Replikation