Amazon RDS for PostgreSQL DB インスタンス間で論理レプリケーションを使用してテーブルを複製する方法を教えてください。

解決策

Amazon RDS for PostgreSQL は、PostgreSQL 10.4 以降での論理レプリケーションをサポートしています。Amazon Aurora PostgreSQL 互換エディションのバージョン 2.2.0 以降では、PostgreSQL 10.6 以降での論理レプリケーションがサポートされています。

詳細については、「Amazon RDS for PostgreSQL で論理レプリケーションを行う」を参照してください。

次の解決策では、2 つのソーステーブルを 2 つのターゲットテーブルに複製します。

論理レプリケーションを有効にする

次の手順を実行します。

1. カスタムパラメータグループを作成し、rds.logical_replication パラメータを 1 に設定します。
   注: rds.logical_replication は静的パラメータであるため、DB インスタンスを再起動する必要があります。DB インスタンスの再起動後、wal_level パラメータは logical に変更されます。
2. パラメータグループを DB インスタンスに関連付けます。
3. 次のクエリを実行し、wal_level が logical に設定されており、rds.logical_replication が有効であることを確認します。

   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)

ソーステーブルを作成し、データを挿入する

PostgreSQL DB インスタンスに接続し、ソースデータベースで次の手順を実行します。

1. 次のコマンドを実行し、ソーステーブルを作成します。

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

2. 次のコマンドを実行し、ソーステーブルにデータを追加します。

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

ソーステーブルのパブリケーションを作成する

ソースデータベースで次の手順を実行します。

1. 次のコマンドを実行し、2 つのテーブルのパブリケーションを作成します。

   CREATE PUBLICATION testpub FOR TABLE reptab1, reptab2;

2. 次のクエリを実行し、パブリケーションの詳細が適切であることを確認します。

   SELECT * FROM pg_publication;

   想定される出力:

     oid   | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot  
   --------+---------+----------+--------------+-----------+-----------+-----------+-------------+------------  
    115069 | testpub |    16395 | f            | t         | t         | t         | t           | f  
   (1 row)

3. 次のクエリを実行し、ソーステーブルがパブリケーションに含まれていることを確認します。

   SELECT * FROM pg_publication;

   想定される出力:

    pubname | schemaname | tablename  
   
   ---------+------------+-----------  
   
    testpub | public | reptab1  
   
    testpub | public | reptab2  
   
   (2 rows)

ターゲットデータベースに接続し、ターゲットテーブルを作成する

ターゲットデータベースで次の手順を実行します。

1. 次のコマンドを実行し、ターゲットテーブルを作成します。

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

   注: ソーステーブルと同じ名前を指定してください。

2. 次のクエリを実行し、データがターゲットテーブルに含まれていないことを確認します。
   テーブル 1:

   SELECT count(*) FROM reptab1;

   想定される出力:

    count  
   -------  
        0  
   (1 row)

   テーブル 2:

   SELECT count(*) FROM reptab2;

   想定される出力:

    count  
   -------  
        0  
   (1 row)

ターゲットデータベースにサブスクリプションを作成する

サブスクリプションを作成する前に、次のコマンドを実行し、データベースログにはユーザー名とパスワードがプレーンテキストで保存されていないことを確認します。

SET log_min_messages to 'PANIC';  
SET log_statement to NONE;

ターゲットデータベースにサブスクリプションを作成するには、次の手順を実行します。

1. サブスクリプションを作成するには、次のコマンドを実行します。

   CREATE SUBSCRIPTION testsub CONNECTION 'host=source RDS/host endpoint port=5432 dbname=source_db_name user=user password=password' PUBLICATION testpub;

   注: source RDS/host endpoint をソース DB インスタンスのエンドポイントに、source_db_name を DB インスタンス名に置き換えてください。user をユーザー名に、password をパスワードに置き換えてください。

2. 次のクエリを実行し、サブスクリプションがアクティブであることを確認します。

   SELECT oid,subname,subenabled,subslotname,subpublications FROM pg_subscription;

   想定される出力:

     oid  | subname | subenabled | subslotname | subpublications  
   -------+---------+------------+-------------+-----------------  
    16434 | testsub | t          | testsub     | {testpub}  
   (1 row)

3. 次のクエリを実行し、データがターゲットテーブルに存在することを確認します。
   テーブル 1:

   SELECT count(*) FROM reptab1;

   想定される出力:

    count  
   -------  
     1000  
   (1 row)

   テーブル 2:

   SELECT count(*) FROM reptab2;

   想定される出力:

    count  
   -------  
       50  
   (1 row)

ソースデータベースのレプリケーションスロットに関する詳細を確認する

ターゲットデータベースでサブスクリプションを作成すると、ターゲットデータベースはソースデータベースにレプリケーションスロットを作成します。

レプリケーションスロットの詳細を確認するには、ソースデータベースで次のクエリを実行します。

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)

ソーステーブルからのレプリケーションをテストする

ソースデータベースで次の手順を実行します。

1. 次のコマンドを実行し、ソーステーブルに行を追加します。

   INSERT INTO reptab1 VALUES(generate_series(1001,2000));  
   INSERT INTO reptab2 SELECT SUBSTR('ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789',((random()*(36-1)+1)::integer),1) FROM generate_series(1,50);

2. 次のクエリを実行し、ソーステーブルのデータが変更されたことを確認します。
   テーブル 1:

   SELECT count(*) FROM reptab1;

   想定される出力:

    count  
   -------  
      2000  
   (1 row)

   テーブル 2:

   SELECT count(*) FROM reptab2;

   想定される出力:

    count  
   -------  
      100  
   (1 row)

ソーステーブルのデータがターゲットテーブルに複製されたことを確認する

ターゲットデータベースで次のクエリを実行し、ソーステーブルのデータがターゲットテーブルに複製されたことを確認します。

テーブル 1:

SELECT count(*) FROM reptab1;

想定される出力:

count  
-------  
   2000  
(1 row)

テーブル 2:

SELECT count(*) FROM reptab2;

想定される出力:

 count  
-------  
   100  
(1 row)

レプリケーションスロットをクリアし、論理レプリケーションを無効にする

レプリケーションの完了後は、論理レプリケーションは必要ありません。スロットをクリアすると無効化されます。レプリケーションスロットが非アクティブの場合、Write-Ahead Logging (WAL) ファイルがソース DB インスタンスに蓄積されます。WAL ファイルによりストレージに空きがなくなり、ダウンタイムが引き起こされる可能性があります。

次の手順を実行します。

1. ターゲットデータベースで次のコマンドを実行し、サブスクリプションを削除します。

   DROP SUBSCRIPTION testsub;

2. ターゲットデータベースで次のクエリを実行し、サブスクリプションが削除されたことを確認します。

   SELECT * FROM pg_subscription;

   想定される出力:

   oid | subdbid | subname | subowner | subenabled | subconninfo | subslotname | subsynccommit | subpublications  
   ----+---------+---------+----------+------------+-------------+-------------+---------------+-----------------  
   (0 rows)

   注: ターゲットデータベースでサブスクリプションを削除すると、ターゲットデータベースはソースデータベースのレプリケーションスロットも削除します。

3. ソースデータベースで次のクエリを実行し、レプリケーションスロットがソースデータベースから削除されたことを確認します。

   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)

4. ソースデータベースで次のコマンドを実行し、パブリケーションを削除します。

   DROP PUBLICATION testpub;

5. ソースデータベースで次のクエリを実行し、パブリケーションが削除されたことを確認します。

   SELECT * FROM pg_publication;  
   SELECT * FROM pg_publication_tables;

   想定される出力:

    oid | pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate | pubviaroot  
   -----+---------+----------+--------------+-----------+-----------+-----------+-------------+------------  
   (0 rows)  
   pubname | schemaname | tablename  
   ---------+------------+-----------  
   (0 rows)

6. DB インスタンスのカスタムパラメータグループで rds.logical_replication パラメータを 0 に設定します。
   注: DB インスタンスを再起動すると、変更が反映されます。

7. max_replication_slots、max_wal_senders、max_logical_replication_workers、max_worker_processes、max_sync_workers_per_subscription の各パラメータを使用状況に応じて確認します。

8. 次のクエリを実行し、非アクティブなレプリケーションスロットの有無とスロットのサイズを確認します。

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

9. (オプション) 次のコマンドを実行してレプリケーションスロットを削除します。

   SELECT pg_drop_replication_slot('Your_slotname_name')

関連情報

レプリケーション (PostgreSQL のウェブサイト)

論理レプリケーション (PostgreSQL のウェブサイト)