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

Logische Decodierung für eine DB-Instance von RDS for 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 for PostgreSQL die test_decoding- und wal2json-Ausgabe-Plugins, die mit PostgreSQL geliefert werden.

1. Erstellen Sie die Quelltabellen

   Connect zur Quelldatenbank in Ihrer RDS for PostgreSQL-DB-Instance her:

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

2. Fügen Sie Daten in die Quelltabellen ein

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

3. Erstellen Sie eine 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)
             

   • Stellen Sie sicher, dass die Quelltabellen der 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:

     
     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 neue Tabellen hinzuzufügen:

     
     ALTER PUBLICATION <publication_name> add table <new_table_name>;
               

4. Connect zur Zieldatenbank her und erstellen Sie Zieltabellen

   • Connect zur Zieldatenbank in der Ziel-DB-Instance her. Erstellen Sie die Zieltabellen mit den gleichen Namen wie die Quelltabellen:

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

   • Stellen Sie sicher, 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 Sie das Abonnement 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 aus 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 Sie den Replikationssteckplatz in der Quelldatenbank

   Durch die Erstellung eines Abonnements in der Zieldatenbank wird ein Replikationsslot in der Quelldatenbank erstellt. Überprüfen Sie die Details des Replikationssteckplatzes, 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)
           

7. Die Replikation wird getestet

   • Testen Sie, ob Datenänderungen in den Quelltabellen in die Zieltabellen repliziert werden, indem Sie 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 neue Einfügungen repliziert werden:

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

8. Aktualisierung 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 Herausgeber abgerufen und die Replikation für Tabellen gestartet, die seit der Erstellung oder letzten Aktualisierung des Abonnements zu den abonnierten Publikationen hinzugefügt wurden.