PolarDB:PolarDB for PostgreSQLと互換性のあるDebeziumコネクタ (Oracleと互換性があります)

PolarDB for PostgreSQL (Compatible with Oracle) とPostgreSQL Community Editionでは、データ型と組み込みオブジェクトの処理にわずかな違いがあります。 このトピックでは、最小限のコード変更で、Debezium PostgreSQLコネクタのコミュニティエディションに基づいてDebezium PolarDBOコネクタを構築する方法について説明します。

Debezium PolarDBOコネクタを構築する

使用上の注意

Debezium PolarDBOコネクタは、論理レプリケーションを使用して、PolarDB for PostgreSQL (Oracle互換) データベースの増分変更をキャプチャします。 次の条件を満たす必要があります。

• wal_levelパラメーターをlogicalに設定して、論理レプリケーションに必要な情報が先行書き込みログ (WAL) レコードに書き込まれるようにします。

  説明

  PolarDBコンソールでwal_levelパラメーターを設定できます。 詳細については、「クラスターパラメーターの設定」トピックの手順セクションをご参照ください。 パラメーターを変更すると、クラスターが再起動します。 作業は慎重に行ってください。

• ALTER TABLE schema.table REPLICA IDENTITY FULL; ステートメントを実行して、サブスクライブするテーブルごとにREPLICA IDENTITYパラメーターをFULLに設定します。 この設定により、関連するテーブル列の前の値が、更新および削除操作のために論理復号プラグインで使用できるようになります。

  説明

  • REPLICA IDENTITYは、PostgreSQL固有のテーブルレベルのパラメーターで、関連するテーブル列の以前の値が更新および削除操作のために論理復号プラグインで使用できるかどうかを決定します。 REPLICA IDENTITY値の詳細については、「レプリカID」をご参照ください。

  • サブスクライブするテーブルのREPLICA IDENTITYパラメーターをFULLに設定すると、テーブルがロックされる可能性があります。 これはビジネスに影響します。 混乱を最小限に抑えるには、この変更を計画し、ビジネスプロセスに基づいて調整します あなたのビジネス要件。 テーブルのREPLICA IDENTITYパラメーターがFULLに設定されているかどうかを確認するには、次のステートメントを実行します。

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

• max_wal_sendersおよびmax_replication_slotsパラメーターの値が、使用するレプリケーションスロットの数およびKafkaジョブで必要なスロットの数よりも大きいことを確認します。

• 特権アカウントまたはLOGINおよびREPLICATION権限を持つ標準アカウントを使用していることを確認してください。 アカウントには、完全なデータクエリをサブスクライブするテーブルに対するSELECT権限が必要です。

• クラスターのプライマリエンドポイントを使用して、PolarDBクラスターに接続します。 クラスターエンドポイントを使用してクラスターに接続する場合、論理レプリケーションはサポートされていません。

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

• plugin.nameパラメーターをpgoutputに設定することを推奨します。 これにより、エンコードされたデータベースの増分解析中のデータの破損や文字化けを防ぎnon-UTF-8。 詳細については、「コミュニティドキュメント」をご参照ください。

例

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

準備

1. カフカをセットアップします。

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

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

      説明

      テストシナリオで読みやすくするために、単一パーティションのトピックを作成することをお勧めします。 実際のビジネスシナリオでは、マルチパーティショントピックを作成することを推奨します。

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

   • Debezium PolarDBOコネクタのJARパッケージをKafka Connectのplugin.pathディレクトリにコピーします。 詳細については、このトピックの「手順」セクションの4番目の手順を参照してください。

     # ${plugin.path} Replace this with the specific 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 (Oracle互換) を設定します。

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

   2. PolarDBクラスターを設定します。 クラスターがDebezium PolarDBOコネクタを使用するための前提条件を満たしていることを確認します。 詳細については、「使用状況のメモ」をご参照ください。

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

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

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

   6. 次のSQL文を実行して、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"
     }
   }

   説明

   デフォルトでは、テーブルごとに個別のトピックが作成されます。 上記の設定は、トピックを集約します。

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

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

   コネクタを追加すると、Kafkaのpg_dbz_eventトピックから完全なデータを取得できます。

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トピックの増分データを照会できます。