ApsaraDB RDS:CDC にレプリケーションスロットを使用する

注意事項

• CDC を有効にしてデータ変更を取得できるのは、プライマリインスタンスのみです。この機能は読み取り専用インスタンスではサポートされていません。

• ApsaraDB RDS for PostgreSQL は、論理レプリケーションスロットのフェールオーバーをサポートしています。プライマリ/セカンダリのスイッチオーバーは CDC に影響しません。詳細については、「論理レプリケーションスロットのフェールオーバー」をご参照ください。

• CDC を有効にする前に、ApsaraDB RDS for PostgreSQL インスタンスのパラメータを変更する必要があります。この操作によりインスタンスが再起動します。サービス中断を回避するため、オフピーク時間に実行してください。

• ApsaraDB RDS for PostgreSQL インスタンスで CDC を使用する際は、以下の点に注意してください。

  • CDC では、WAL ログのためにより多くのストレージ容量が必要になります。データ取得が異常になった場合や停止した場合、インスタンスは WAL ログを自動的にクリーンアップしません。これによりログが蓄積し、利用可能なディスク容量をすべて消費する可能性があり、インスタンスがロックされる恐れがあります。ロックされたインスタンスは読み取り専用になり、書き込み操作を拒否します。

  • 古いデータ行を長期間保持すると、トランザクション ID 周回問題が発生し、インスタンスのすべての書き込み操作がブロックされる可能性があります。エラーログには次のようなメッセージが記録される場合があります。

    "HINT: Close open transactions soon to avoid wraparound problems. You might also need to commit or roll back old prepared transactions, or drop stale replication slots."
    "WARNING: oldest xmin is far in the past."

  レプリケーションスロットを手動で削除することで、PostgreSQL カーネルが古い WAL ログとデータ行を自動的にクリーンアップできるようになります。詳細については、「CDC の無効化」をご参照ください。

  説明

  インスタンスの WAL ログサイズとディスク使用量を定期的に監視し、関連するアラートを設定してください。詳細については、「拡張モニタリングデータの表示」および「アラートルールの管理」をご参照ください。

CDC の有効化

ステップ 1：テストデータベースの作成

1. RDSインスタンスにアクセスし、上部のリージョンを選択し、対象のRDSインスタンスのIDをクリックします。

2. 左側のナビゲーションペインで、データベース管理 を選択します。

3. データベースを作成する をクリックします。 testdb データベースが作成されると、そのステータスは 実行中 になり、文字セットは UTF8 になります。

   説明

   この例では、[testdb]という名前のデータベースが作成されます。詳細については、「データベースを作成する」をご参照ください。

ステップ 2：テストアカウントの作成と権限の設定

1. 左側のナビゲーションペインで、アカウント管理 を選択します。

2. アカウントを作成する をクリックして、特権アカウント ([db_admin]) と CDC 用の標準アカウント ([cdc_user]) を作成します。

   説明

   このトピックで使用されているアカウント名は、デモンストレーション目的のみです。カスタム名を指定できます。アカウントの作成方法の詳細については、「アカウントの作成」をご参照ください。

3. db_admin アカウントを使用して ApsaraDB RDS for PostgreSQL インスタンスに接続します。

   psql -h <ApsaraDB RDS for PostgreSQL インスタンスのエンドポイント> -p 5432 -U db_admin -d testdb

   説明

   インスタンスのエンドポイントを取得する方法の詳細については、「エンドポイントとポート番号の表示または変更」をご参照ください。

4. 次のコマンドを実行して、[cdc_user] アカウントをレプリケーションロールに追加します：

   ALTER USER cdc_user WITH REPLICATION;

   次のコマンドを実行して結果を確認できます。

   SELECT rolreplication FROM  pg_roles WHERE rolname='cdc_user';

   出力例：

    rolreplication
   ----------------
    t
   (1 行)

5. 次のコマンドを実行して、[cdc_user] アカウントに権限を付与します：

   GRANT SELECT ON ALL TABLES IN SCHEMA PUBLIC to cdc_user;

ステップ 3：インスタンスパラメータの変更

1. 次のコマンドを実行して、現在のパラメータ設定を確認します。

   SELECT name,
          setting,
          short_desc,
          source
   FROM pg_settings
   WHERE name ='wal_level';

   出力例：

            name          | setting |                               short_desc                                |       source
   -----------------------+---------+-------------------------------------------------------------------------+--------------------
    wal_level             | replica | Sets the level of information written to the WAL.                       | configuration file
   (1 行)

   wal_level パラメータは、PostgreSQL が WAL ログに書き込む情報量を制御します。デフォルト値は replica です。このパラメータはサーバー起動時にのみ設定できます。有効な値：

   • minimal： クラッシュまたは即時シャットダウンからの復旧に必要な情報のみを書き込みます。このレベルでは、ベースバックアップと WAL ログからデータベースを復元することはできません。

   • replica： WAL アーカイブとレプリケーションをサポートするのに十分な情報を書き込みます。スタンバイサーバーでの読み取り専用クエリの実行も含まれます。

   • logical： 論理デコーディングに必要な情報を追加します。

2. RDSインスタンスにアクセスし、上部のリージョンを選択し、対象のRDSインスタンスのIDをクリックします。

3. 左側のナビゲーションペインで、パラメーターの設定 を選択します。

4. [wal_level] パラメーターを [logical] に設定します。

   説明

   • インスタンスパラメータの変更方法の詳細については、「ApsaraDB RDS for PostgreSQL インスタンスのパラメータの変更」をご参照ください。

   • パラメータ変更を送信すると、ApsaraDB RDS for PostgreSQL インスタンスが再起動します。サービス中断を回避するため、オフピーク時間に実行してください。

ステップ 4：論理レプリケーションスロットの作成

1. [db_admin] アカウントを使用して ApsaraDB RDS for PostgreSQL インスタンスに接続します。

   psql -h <ApsaraDB RDS for PostgreSQL インスタンスのエンドポイント> -p 5432 -U db_admin -d testdb

2. 次のコマンドを実行し、test_decoding 出力プラグインを使用して cdc_replication_slot という名前のレプリケーションスロットを作成します。

   SELECT pg_create_logical_replication_slot('cdc_replication_slot', 'test_decoding');

   説明

   • スロット名 cdc_replication_slot はデモンストレーション用です。カスタム名を指定できます。

   • PostgreSQL は test_decoding 出力プラグインをネイティブで提供しているため、変更は不要です。

   出力例：

    pg_create_logical_replication_slot
   ------------------------------------
    (cdc_replication_slot,1/14003428)
   (1 行)

   次のコマンドを実行して、スロットが作成されたかどうかを確認できます。

   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  | two_phase
   ----------------------+---------------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+----------------+-----------
    cdc_replication_slot | test_decoding | logical   |  18822 | testdb   | f         | f      |            |      |        22356 | 1/140033F0  | 1/14003428          | reserved   |                | f
   (1 行)

ステップ 5：テストデータの作成

CREATE TABLE public.tb_test(
    id int NOT NULL PRIMARY KEY
);
ALTER TABLE public.tb_test ADD name varchar(1) NULL;
INSERT INTO public.tb_test SELECT 1, 'A';

ステップ 6：スロットからのデータ読み取り

1. [cdc_user] アカウントを使用して ApsaraDB RDS for PostgreSQL インスタンスに接続します。

   psql -h <ApsaraDB RDS for PostgreSQL インスタンスのエンドポイント> -p 5432 -U cdc_user -d testdb

2. 次のコマンドを実行して、レプリケーションスロットから変更を読み取ります。

   SELECT * FROM pg_logical_slot_peek_changes('cdc_replication_slot', null, null);

   出力例：

       lsn     |  xid  |                                  data
   ------------+-------+-------------------------------------------------------------------------
    1/14003D90 | 22376 | BEGIN 22376
    1/1401DDE8 | 22376 | COMMIT 22376
    1/1401DDE8 | 22377 | BEGIN 22377
    1/1401E100 | 22377 | COMMIT 22377
    1/1401E2A8 | 22382 | BEGIN 22382
    1/1401E2A8 | 22382 | table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
    1/1401E3C0 | 22382 | COMMIT 22382
   (7 行)

ステップ 7：データ変更の消費

1. \q コマンドを実行して、データベース接続を終了します。

2. 次のコマンドを実行して、データ変更を消費します。

   説明

   pg_recvlogical コマンドは、postgres ユーザーとして実行する必要があります。su - postgres コマンドを実行してユーザーを切り替えることができます。-bash: pg_recvlogical: command not found エラーが発生した場合は、解決策について よくある質問 セクションをご参照ください。

   pg_recvlogical -h <ApsaraDB RDS for PostgreSQL インスタンスのエンドポイント> -U <特権アカウント> -d <テストデータベース> --create-slot --if-not-exists --slot=cdc_replication_slot --plugin=test_decoding --start -f -

   コマンド例：

   pg_recvlogical -h pgm-*****.pgsql.singapore.rds.aliyuncs.com -U db_admin -d testdb --create-slot --if-not-exists --slot=cdc_replication_slot --plugin=test_decoding --start -f -

   出力例：

   BEGIN 22376
   COMMIT 22376
   BEGIN 22377
   COMMIT 22377
   BEGIN 22382
   table public.tb_test: INSERT: id[integer]:1 name[character varying]:'A'
   COMMIT 22382

CDC の無効化

データ取得が異常になった場合や停止した場合、WAL ログが蓄積し、利用可能なディスク容量をすべて消費して、最終的にインスタンスがロックされる可能性があります。これを解決するには、非アクティブなレプリケーションスロットを手動で削除します。これにより、PostgreSQL カーネルが古い WAL ログをクリーンアップできるようになります。

よくある質問

• Q: データ消費時に -bash: pg_recvlogical: command not found というエラーが表示された場合はどうすればよいですか？

  A: pg_recvlogical は、PostgreSQL のネイティブな論理デコーディングツールであり、デフォルトの test_decoding 出力プラグインを使用します。このプラグインは、PostgreSQL のソースコードパッケージの contrib/test_decoding ディレクトリにあります。PostgreSQL をソースコードからコンパイルしてインストールします。インストール後、pg_recvlogical ツールは、クライアントのインストールパスの /bin ディレクトリに配置されます。PostgreSQL をソースからインストールする方法の詳細については、ソースコードからのインストールをご参照ください。

• Q： レプリケーションスロットを手動で削除しましたが、システムが WAL ログを自動的にクリーンアップせず、ディスク容量を消費し続けています。どうすればよいですか。

  A: wal_keep_segments パラメーターをデフォルト値の 128 に変更すると、保持されるログファイルの数を減らすことができます。インスタンスパラメーターの変更方法の詳細については、「ApsaraDB RDS for PostgreSQL インスタンスのパラメーターを変更する」をご参照ください。