PolarDB:PolarDB for PostgreSQL (Compatible with Oracle) 向け Debezium コネクター

PolarDB for PostgreSQL (Compatible with Oracle) とコミュニティ版 PostgreSQL の違いは、少数のデータ型と組み込みオブジェクトの処理方法のみであるため、本トピックでは、コミュニティの Debezium PostgreSQL コネクターを最小限のコード変更で適応させ、PolarDB for PostgreSQL (Compatible with Oracle) をサポートする Debezium コネクターをビルドする方法について説明します。

Debezium PolarDBO コネクターのビルド

使用方法

Debezium PolarDBO コネクターは、論理レプリケーションを使用して PolarDB for PostgreSQL (Compatible with Oracle) データベースから増分変更を読み取ります。コネクターを使用する前に、以下の条件を満たす必要があります：

• wal_level クラスターパラメーターを logical に設定する必要があります。この設定により、論理レプリケーションに必要な情報が先行書き込みログ (WAL) に追加されます。

  説明

  wal_level クラスターパラメーターはコンソールで設定できます。詳細については、「クラスターパラメーターの設定」をご参照ください。このパラメーターを変更するとクラスターが再起動します。ビジネスニーズに応じて、この操作を慎重に計画してください。

• ALTER TABLE schema.table REPLICA IDENTITY FULL; コマンドを実行して、各サブスクライブされたテーブルの REPLICA IDENTITY を FULL に設定します。これにより、挿入および更新イベントにすべての列の以前の値が含まれるようになり、データの一貫性が保証されます。

  説明

  • REPLICA IDENTITY は PostgreSQL 固有のテーブルレベルの設定です。これにより、INSERT および UPDATE イベント中に論理デコーディングプラグインが関連するテーブル列の以前の値を含めるかどうかが決まります。REPLICA IDENTITY の値の詳細については、「REPLICA IDENTITY」をご参照ください。

  • サブスクライブされたテーブルの REPLICA IDENTITY を FULL に設定すると、テーブルロックが必要になる場合があり、サービスに影響を与える可能性があります。ビジネスニーズに応じて、この操作を計画してください。次のコマンドを実行して、現在の設定が FULL であるかどうかを確認できます：

    SELECT relreplident = 'f' FROM pg_class WHERE relname = 'tablename';

• max_wal_senders および max_replication_slots パラメーターの値は、現在使用中のレプリケーションスロットと Kafka ジョブで必要なレプリケーションスロットの合計よりも大きい必要があります。

• 特権アカウント、または LOGIN と REPLICATION の両方の権限を持つ標準アカウントを使用します。初期スナップショットのために、アカウントはすべてのサブスクライブされたテーブルに対する SELECT 権限も持っている必要があります。

• PolarDB クラスターの プライマリエンドポイント にのみ接続してください。論理レプリケーションは クラスターエンドポイント ではサポートされていません。

• connector.class パラメーターを io.debezium.connector.postgresql.PolarDBOConnector に設定します。

• plugin.name パラメーターを pgoutput に設定することを推奨します。そうしないと、UTF-8 エンコーディングを使用しないデータベースで増分解析を行うと、テキストが文字化けする可能性があります。詳細については、「コミュニティドキュメント」をご参照ください。

例

次の例では、Debezium PolarDBO コネクターを使用して、PolarDB for PostgreSQL Oracle 互換性 2.0 クラスター内の dbz_db データベースから t1 テーブルと t2 テーブルを Kafka メッセージキューに同期する方法について説明します。

前提条件

1. Kafka のセットアップ

   1. Kafka インスタンス をデプロイし、Kafka Connect ホスト からアクセスできることを確認します。ApsaraMQ for Kafka を使用することもできます。詳細については、「クイックスタート」をご参照ください。

   2. メッセージを受信するために、Kafka インスタンスに pg_dbz_event という名前のトピックを作成します。

      説明

      テスト目的では、表示を容易にするために単一 パーティション のトピックを作成できます。本番環境では、複数パーティションのトピックを作成してください。

2. ポート 8083 で、Kafka Connect をローカルで 分散モード で起動します。

   • Debezium PolarDBO コネクターの JAR パッケージを Kafka Connect の plugin.path ディレクトリにコピーします。

     # ${plugin.path} を実際のパスに置き換えます。
     mkdir ${plugin.path}/debezium-connector-polardbo
     cp debezium-connector-postgres-polardbo-v1.0-2.6.2.Final.jar ${plugin.path}/debezium-connector-polardbo

3. PolarDB for PostgreSQL (Compatible with Oracle) のセットアップ

   1. PolarDB クラスター購入ページで、PolarDB for PostgreSQL (Compatible with Oracle) 2.0 クラスターを購入します。

   2. 使用方法セクションに記載されているすべての前提条件を満たすように PolarDB クラスターを設定します。

   3. 特権アカウントを作成します。詳細については、「アカウントの作成」をご参照ください。

   4. クラスターの プライマリエンドポイント を取得します。詳細については、「接続エンドポイントの表示」をご参照ください。PolarDB クラスターと Kafka Connect インスタンス のアドレスが同じ アベイラビリティーゾーン にある場合は、プライベートエンドポイント を使用できます。それ以外の場合は、パブリックエンドポイント をリクエストする必要があります。Kafka Connect インスタンス のアドレスを PolarDB クラスターホワイトリスト に追加します。詳細については、「クラスターホワイトリストの設定」をご参照ください。

   5. コンソールで、 dbz_db という名前のデータベースを作成します。詳細については、「データベースの作成」をご参照ください。

   6. 次のステートメントを実行して、dbz_db データベースに t1 および t2 テーブルを作成し、データを挿入します。

      CREATE TABLE public.t1 (a int PRIMARY KEY, b text, c TIMESTAMP);
      ALTER TABLE public.t1 REPLICA IDENTITY FULL;
      INSERT INTO public.t1(a, b, c) VALUES(1, 'a', now());
      CREATE TABLE public.t2 (a int PRIMARY KEY, b text, c DATE);
      ALTER TABLE public.t2 REPLICA IDENTITY FULL;
      INSERT INTO public.t2(a, b, c) VALUES(1, 'a', now());

テスト

1. config/postgresql-connector.json という名前の設定ファイルを作成します。パラメーターの説明については、「コミュニティドキュメント」をご参照ください。

   {
     "name": "dbz-polardb",
     "config": {
       "connector.class": "io.debezium.connector.postgresql.PolarDBOConnector",
       "database.hostname": "<yourHostname>", 
       "database.port": "<yourPort>", 
       "database.user": "<yourUserName>", 
       "database.password": "<yourPassWord>", 
       "database.dbname" : "dbz_db",
       "plugin.name": "pgoutput",
       "slot.name": "dbz_polardb",
       "table.include.list": "public.t1,public.t2",
       "topic.prefix": "polardb",
       "transforms": "Combine",
       "transforms.Combine.type": "io.debezium.transforms.ByLogicalTableRouter",
       "transforms.Combine.topic.regex": "(.*)",
       "transforms.Combine.topic.replacement": "pg_dbz_event"
     }
   }

   説明

   デフォルトでは、Debezium はテーブルごとに 1 つのトピックを作成します。この設定では、すべての変更イベントが単一のトピックにルーティングされます。

2. コネクターを追加します。

   curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" 'http://localhost:8083/connectors' -d @config/postgresql-connector.json

   コネクターが追加されると、完全なデータが Kafka トピックで利用できるようになります。

   Kafka インスタンスの [メッセージクエリ] タブで、クエリ方法を [オフセットによるクエリ] に設定し、パーティション 0 を選択して、[開始オフセット] を 0 に設定します。次に、[クエリ] をクリックします。結果には、2 つの Debezium CDC メッセージ (オフセット 0 と 1) が表示されます。キーには、それぞれ __dbz__physicalTableIdentifier の値が polardb.public.t1 と polardb.public.t2 として含まれています。これにより、完全なデータスナップショットが Kafka に同期されたことが確認できます。

3. PolarDB クラスターの dbz_db データベースで、次の DML ステートメントを実行します：

   INSERT INTO public.t1(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t1 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t1 WHERE a = 2;
   INSERT INTO public.t1(a, b, c) VALUES(4, 'd', now());
   INSERT INTO public.t2(a, b, c) VALUES(2, 'b', now());
   UPDATE public.t2 SET b = 'c' WHERE a = 1;
   DELETE FROM public.t2 WHERE a = 2;
   INSERT INTO public.t2(a, b, c) VALUES(4, 'd', now());

   増分データが Kafka トピックで利用できるようになります。

   メッセージクエリページで、 [クエリ方法] を [オフセットによるクエリ] に設定し、対象のパーティションと開始オフセットを選択して、 [クエリ] をクリックします。結果には、Debezium が t1 および t2 テーブルに対する DML 操作を CDC メッセージとしてキャプチャしたことが示されます。各メッセージのキーには、テーブル識別子 (例：polardb.public.t1) と主キーの値が含まれています。メッセージの値には、before および after フィールドを含む、Debezium 形式の変更イベントが含まれています。DELETE 操作の場合、after フィールドは null になり、before フィールドに削除された行のデータが格納されます。