Ausführen der logischen Replikation für Amazon RDS für PostgreSQL

Logische Decodierung für eine DB-Instance von RDS für PostgreSQL aktivieren

1. Stellen Sie sicher, dass das Benutzerkonto, das Sie verwenden, folgende Rollen hat:

   • Die rds_superuser-Rolle, damit Sie die logische Replikation aktivieren können

   • Die rds_replication-Rolle erteilt Berechtigungen zur Verwaltung von logischen Slots und zum Streamen von Daten mithilfe von logischen Slots

2. Setzen Sie den statischen Parameter rds.logical_replication auf 1. Setzen Sie bei der Anwendung dieses Parameters auch die Parameter wal_level, max_wal_senders, max_replication_slots und max_connections fest. Diese Änderungen an den Parametern können die WAL-Generierung steigern, legen Sie daher den Parameter rds.logical_replication nur dann fest, wenn Sie logische Slots verwenden.

3. Starten Sie die DB-Instance neu, damit der statische Parameter rds.logical_replication in Kraft tritt.

4. Erstellen Sie einen logischen Replikationsslot wie im nächsten Abschnitt erläutert. Für diesen Prozess ist es erforderlich, dass Sie ein Decodier-Plugin angeben. Derzeit unterstützt RDS für PostgreSQL die test_decoding- und wal2json-Ausgabe-Plugins, die mit PostgreSQL geliefert werden.

1. Erstellen der Quelltabellen

   Stellen Sie eine Verbindung mit Ihrer Quelldatenbank in der DB-Instance von RDS für PostgreSQL her.

   
   source=> CREATE TABLE testtab (slno int primary key);
   CREATE TABLE
           

2. Einfügen der Daten in die Quelltabellen

   
   source=> INSERT INTO testtab VALUES (generate_series(1,1000));
   INSERT 0 1000
           

3. Erstellen einer Publikation für Quelltabellen

   • Erstellen Sie eine Publikation für die Quelltabellen:

     
     source=> CREATE PUBLICATION testpub FOR TABLE testtab;
     CREATE PUBLICATION
             

   • Verwenden Sie eine SELECT-Abfrage, um die Details der erstellten Publikation zu überprüfen:

     
     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)
             

   • Vergewissern Sie sich, dass die Quelltabellen zur Publikation hinzugefügt wurden:

     
     source=> SELECT * FROM pg_publication_tables; 
     pubname | schemaname | tablename
     ---------+------------+-----------
      testpub | public     | testtab
     (1 rows)
             

   • Um alle Tabellen in einer Datenbank zu replizieren, verwenden Sie Folgendes:

     
     CREATE PUBLICATION testpub FOR ALL TABLES;
               

   • Wenn die Publikation bereits für eine einzelne Tabelle erstellt wurde und Sie eine neue Tabelle hinzufügen müssen, können Sie die folgende Abfrage ausführen, um der vorhandenen Publikation beliebige neue Tabellen hinzuzufügen:

     
     ALTER PUBLICATION <publication_name> add table <new_table_name>;
               

4. Herstellen einer Verbindung mit der Zieldatenbank und Erstellen von Zieltabellen

   • Stellen Sie eine Verbindung mit der Zieldatenbank in der Ziel-DB-Instance her. Erstellen Sie die Zieltabellen mit denselben Namen wie die Quelltabellen:

     
     target=> CREATE TABLE testtab (slno int primary key);
     CREATE TABLE
     

   • Vergewissern Sie sich, dass in den Zieltabellen keine Daten vorhanden sind, indem Sie eine SELECT-Abfrage für die Zieltabellen ausführen:

         
     target=> SELECT count(*) FROM testtab;
      count
     -------
          0
     (1 row)
             

5. Erstellen und Überprüfen des Abonnements in der Zieldatenbank

   • Erstellen Sie das Abonnement in der Zieldatenbank:

     
     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
             

   • Verwenden Sie eine SELECT-Abfrage, um zu überprüfen, ob das Abonnement aktiviert ist:

     
     target=> SELECT oid, subname, subenabled, subslotname, subpublications FROM pg_subscription;
       oid  | subname | subenabled | subslotname | subpublications
     -------+---------+------------+-------------+-----------------
      16434 | testsub | t          | testsub     | {testpub}
     (1 row)
             

   • Wenn das Abonnement erstellt wird, werden alle Daten von den Quelltabellen in die Zieltabellen geladen. Führen Sie eine SELECT-Abfrage für die Zieltabellen aus, um zu überprüfen, ob die ursprünglichen Daten geladen werden:

     
     target=> SELECT count(*) FROM testtab;
      count
     -------
       1000
     (1 row)
             

6. Überprüfen des Replikations-Slots in der Quelldatenbank

   Durch die Erstellung eines Abonnements in der Zieldatenbank wird ein Replikations-Slot in der Quelldatenbank erstellt. Überprüfen Sie die Details des Replikations-Slots, indem Sie die folgende SELECT-Abfrage für die 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)
           

7. Testen der Replikation

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

     
     source=> INSERT INTO testtab VALUES(generate_series(1001,2000));
     INSERT 0 1000
     
     source=> SELECT count(*) FROM testtab; 
      count
     -------
       2000
     (1 row)
           

   • Überprüfen Sie die Anzahl der Zeilen in den Zieltabellen, um sicherzustellen, dass neu eingefügte Daten repliziert werden:

     
     target=> SELECT count(*) FROM testtab;
      count
     -------
       2000
     (1 row)
           

8. Aktualisieren des Abonnements nach dem Hinzufügen von Tabellen

   • Wenn Sie einer vorhandenen Publikation neue Tabellen hinzufügen, muss das Abonnement unbedingt aktualisiert werden, damit die Änderungen wirksam werden:

     
     ALTER SUBSCRIPTION <subscription_name> REFRESH PUBLICATION;
           

   • Mit diesem Befehl werden fehlende Tabelleninformationen vom Publisher abgerufen und die Replikation für Tabellen gestartet, die seit der Erstellung oder letzten Aktualisierung des Abonnements zu den abonnierten Publikationen hinzugefügt wurden.