汎用 PostgreSQL ソース コネクタを作成する

このドキュメントでは、汎用 PostgreSQL ソースコネクタを作成する方法について説明します。

汎用 PostgreSQL ソースコネクタは、 Debezium PostgreSQL コネクタのインスタンスです。 PostgreSQL データベースから行レベルの変更を読み取り、Managed Service for Apache Kafka クラスタのトピックに書き込みます。

このコネクタのユースケースは次のとおりです。

  • 行レベルのデータベース変更をリアルタイムでモニタリングする。
  • データベース変更イベントをイベント ドリブン アーキテクチャに統合する。
  • 行の挿入や削除などのデータベース イベントに対応する。
  • データベースの変更を他のシステムにコピーする。
  • PostgreSQL テーブルを複製または復元する。

始める前に

汎用 PostgreSQL ソースコネクタを作成する前に、次のものがあることを確認してください。

  • PostgreSQL データベース。

  • Kafka クラスタに関連付けられた Connect クラスタ 。

  • データベース パスワードを保存する Secret Manager シークレットを作成します。構成でデータベース SSL を使用する場合は、データベース SSL パスワードのシークレットも作成します。シークレットを使用して Connect クラスタを構成します。詳細については、 Secret Manager リソースをご覧ください。

必要なロールと権限

コネクタの作成に必要な権限を取得するには、プロジェクトに対するManaged Kafka コネクタ編集者 （roles/managedkafka.connectorEditor）IAM ロールの付与を管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。

この事前定義ロールには コネクタの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。

必要な権限

コネクタを作成するには、次の権限が必要です。

  • コネクタを作成する: managedkafka.connectors.create

カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。

Secret Manager シークレットにアクセスする権限を付与する

Managed Kafka サービス アカウントには、Secret Manager に保存されているシークレットを表示してアクセスする権限が必要です。 次の IAM ロールをサービス アカウントに付与します。

  • Secret Manager 閲覧者 （roles/secretmanager.viewer）
  • Secret Manager のシークレット アクセサー （roles/secretmanager.secretAccessor）

Managed Kafka サービス アカウントの形式は service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com です。ここで、PROJECT_NUMBER は Connect クラスタのプロジェクト番号です。

Connect クラスタが Managed Service for Apache Kafka クラスタとは異なるプロジェクトにある場合は、 別のプロジェクトに Connect クラスタを作成するをご覧ください。

PostgreSQL データベースの構成

コネクタがデータベースからデータ変更イベントを読み取れるようにするには、次の設定を構成します。

  1. サーバーの wal_level を logicalに設定します。

    ALTER SYSTEM SET wal_level = logical;

    設定を適用するには、サーバーを再起動します。

  2. コネクタが PostgreSQL に対して認証を行うためのデータベース ユーザーを作成します。データベース ユーザーはレプリケーション ロールである必要があります。これにより、レプリケーション モードでサーバーに接続できます。

    CREATE ROLE ROLE_NAME WITH REPLICATION LOGIN PASSWORD 'ROLE_PASSWORD';

    次のように置き換えます。

    • ROLE_NAME: ユーザーの名前（debezium_user など）。
    • ROLE_PASSWORD: ユーザーのパスワード。

  3. キャプチャするテーブルの パブリケーション を作成します。コネクタはパブリケーションをサブスクライブして、データ変更イベントを受信します。

    CREATE PUBLICATION dbz_publication FOR TABLE "SCHEMA_NAME"."TABLE_NAME";

    次のように置き換えます。

    • SCHEMA_NAME: テーブルのスキーマ。

    • TABLE_NAME: テーブル名。

    名前が特殊文字や大文字を含む場合は、構文エラーを回避するため、スキーマ名とテーブル名を二重引用符で囲むことをおすすめします。

    または、データベース内のすべてのテーブルの変更を複製するパブリケーションを作成することもできます。

    CREATE PUBLICATION dbz_publication FOR ALL TABLES;

    コネクタの publication.autocreate.mode 設定に応じて、パブリケーションを手動で作成することも、コネクタに自動的に作成させることもできます。 詳細については、パブリケーション モードをご覧ください。

  4. テーブルごとに、データベース ユーザーにテーブルに対する SELECT 権限を付与します。

    GRANT SELECT ON TABLE "SCHEMA_NAME"."TABLE_NAME" TO ROLE_NAME;

    または、スキーマ内のすべてのテーブルに対して選択権限を付与することもできます。

    GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

  5. テーブルごとに、データベース ユーザーにテーブル スキーマに対する USAGE 権限を付与します。テーブルがデフォルトの public スキーマにある場合は、この手順をスキップできます。

    GRANT USAGE ON SCHEMA "SCHEMA_NAME" TO ROLE_NAME;

汎用 PostgreSQL ソースコネクタを作成する

汎用 PostgreSQL ソースコネクタを作成する手順は次のとおりです。

コネクタが初期化されると、次のアクションが実行されます。

  1. データベースの初期スナップショットを作成します。
  2. 行があるテーブルごとに Kafka トピックを作成します。
  3. データベースの行ごとに、対応するトピックに変更イベントを送信します。

コネクタの実行中は、トピックに変更イベントが送信され続けます。初期スナップショットの詳細については、Debezium ドキュメントの スナップショット をご覧ください。

コンソール

  1. コンソール Google Cloud で、[Connect クラスタ] ページに移動します。

    [Connect クラスタ] に移動

  2. コネクタを作成する Connect クラスタをクリックします。

  3. [コネクタを作成] をクリックします。

  4. コネクタ名に文字列を入力します。

    コネクタの命名方法のガイドラインについては、 Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。

  5. [**コネクタ プラグイン**] で [**汎用 PostgreSQL ソース**] を選択します。

  6. [データベース ホスト名] フィールドに、PostgreSQL サーバーのホスト名または IP アドレスを入力します。

  7. [データベース名] フィールドに、データベースの名前を入力します。

  8. [データベース ユーザー] フィールドに、レプリカ ロールの名前を入力します。コネクタは、このロールを使用して PostgreSQL サーバーに対して認証を行います。

  9. [トピック プレフィックス] フィールドに、Kafka トピック名に使用するプレフィックスを入力します。

  10. [シークレット] リストで、データベース パスワードを含むシークレットを選択します。

  11. 省略可: [構成] ボックスで、構成プロパティを追加するか、デフォルトのプロパティを編集します。詳細については、 コネクタを構成するをご覧ください。

  12. 省略可: [タスクの再起動ポリシー] を選択します。詳細については、 タスクの再起動ポリシーをご覧ください。

  13. [作成] をクリックします。

gcloud

  1. コンソール Google Cloud で Cloud Shell をアクティブにします。

    Cloud Shell をアクティブにする

    コンソールの下部にある Google Cloud Cloud Shell セッションが開始し、コマンドライン プロンプトが表示されます。Cloud Shell はシェル環境です 。Google Cloud CLI がすでにインストールされており、現在のプロジェクトの値もすでに設定されています 。セッションが初期化されるまで数秒かかることがあります。

  2. gcloud managed-kafka connectors create コマンドを実行します。

    gcloud managed-kafka connectors create CONNECTOR_ID \
    --location=LOCATION \
    --connect-cluster=CONNECT_CLUSTER_ID \
    --config-file=CONFIG_FILE

    次のように置き換えます。

    • CONNECTOR_ID: コネクタの ID または名前。 コネクタの命名方法のガイドラインについては、 Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。 コネクタの名前は変更できません。

    • LOCATION: コネクタを作成するロケーション。これは、 Connect クラスタを作成したロケーションと同じにする必要があります。

    • CONNECT_CLUSTER_ID: コネクタが作成される Connect クラスタの ID。

    • CONFIG_FILE: コネクタの YAML 構成ファイルのパス。

    汎用 PostgreSQL ソースコネクタの構成ファイルの例を次に示します。

    connector.class: io.debezium.connector.postgresql.PostgresConnector
database.dbname: DATABASE_NAME
database.hostname: HOSTNAME
database.password: CREDENTIALS
database.user: DATABASE_USER
key.converter: org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable: "false"
plugin.name: pgoutput
topic.prefix: TOPIC_PREFIX
value.converter: org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable: "true"

    次のように置き換えます。

    • HOSTNAME: 読み取る PostgreSQL データベースのホスト名。

    • DATABASE_NAME: 読み取る PostgreSQL データベースの名前。

    • DATABASE_USER: データベースの認証時に使用する PostgreSQL データベース ユーザー。

    • CREDENTIALS: データベース パスワードを含む Secret Manager シークレットへのパス。次の形式でシークレットを指定します。

      ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}

    • TOPIC_PREFIX: Kafka トピック名に使用するプレフィックス。

コネクタを構成する

このセクションでは、コネクタに設定できる構成プロパティについて説明します。完全なリストについては、Debezium ドキュメントの PostgreSQL 用 Debezium コネクタ をご覧ください。

パスワードと SSL パスワードの構成

database.password 構成と database.sslpassword 構成では、シークレット パスのみがサポートされています。バックエンドでは、これらの構成が次の形式で使用されることを想定しています: ${directory:/var/secrets:PROJECT_ID-SECRET_NAME-SECRET_VERSION}。

IP アドレスの種類

driver.ipTypes プロパティは、コネクタがデータベースへの接続に使用する IP アドレスのタイプを指定します。

  • PRIVATE: プライベート IP
  • PSC: Private Service Connect
  • PUBLIC: パブリック IP

driver.ipTypes プロパティには、優先順位で並べられた IP タイプのカンマ区切りのリストが含まれます（例: driver.ipTypes=PRIVATE,PUBLIC）。

パブリケーション モード

汎用 PostgreSQL ソースコネクタは、データベース内の パブリケーション から変更イベントをストリーミングします。パブリケーションは手動で作成することも、コネクタに自動的に作成させることもできます。

publication.autocreate.mode 設定は、コネクタがパブリケーションを作成する方法と作成するかどうかを指定します。

  • filtered。パブリケーションが存在しない場合、コネクタはキャプチャされたテーブルのみを含む新しいパブリケーションを作成します。データベース ユーザーには、データベースに対する CREATE 権限があり、含まれているテーブルのオーナーである必要があります。

    パブリケーションがすでに存在する場合、コネクタはキャプチャされたテーブルを含めるように変更します。既存のパブリケーションを変更するには、データベース ユーザーがパブリケーションのオーナーであり、含まれているテーブルのオーナーである必要があります。

  • all_tables。パブリケーションが存在しない場合、コネクタは FOR ALL TABLES パラメータを使用して新しいパブリケーションを作成します。データベース ユーザーはスーパー ユーザーである必要があります。

    スーパー ユーザー ロールはデータベース内のすべての権限チェックをバイパスするため、データベース ユーザーに SUPERUSER を付与することをおすすめしません。代わりに、パブリケーションを手動で作成するか、publication.autocreate.mode=filtered を設定します。

  • disabled。パブリケーションが存在しない場合は、エラーが発生します。コネクタは新しいパブリケーションを作成しません。

デフォルト値は all_tables です。

パブリケーション名

デフォルトでは、コネクタは dbz_publication という名前のパブリケーションからストリーミングしようとします。 別のパブリケーションを指定するには、構成に publication.name=PUBLICATION_NAME を追加します。ここで PUBLICATION_NAME はパブリケーション名です。例: publication.name=my_publication。

レプリケーション スロット

PostgreSQL は、データベース テーブルの変更をストリーミングするためにレプリケーション スロットを使用します。 デフォルトでは、コネクタは debezium という名前のレプリケーション スロットを作成します。別のスロット名を使用するには、slot.name プロパティを設定します。

同じデータベースに対してコネクタのインスタンスを 2 つ作成する場合は、コネクタごとに一意のスロット名を指定する必要があります。

デフォルトでは、コネクタはデータ損失を防ぐために slot.drop.on.stop プロパティを false に設定します。コネクタを完全に削除する場合は、コネクタが使用していたレプリケーション スロットを手動で削除する必要があります。レプリケーション スロット名は、 別の構成にしない限りslot.name、デフォルトで debeziumになります。

ソース PostgreSQL データベース サーバーの WAL ディスク使用量をモニタリングするアラートを設定し、未使用のレプリケーション スロットを削除することをおすすめします。

テーブルのフィルタ

デフォルトでは、コネクタはデータベース内のすべての非システム テーブルから変更データをキャプチャします。キャプチャするテーブルをフィルタリングするには、次の設定を 1 つ以上指定します。

  • schema.include.list。含めるスキーマのリスト。
  • schema.exclude.list。除外するスキーマのリスト。 と一緒に使用することはできませんschema.include.list。
  • table.include.list。含めるテーブルのリスト。
  • table.exclude.list。除外するテーブルのリスト。 と一緒に使用することはできませんtable.include.list.

トピック名

デフォルトでは、コネクタは次の命名規則で Kafka トピックを作成します。topic_prefix.schema.table_name。ここで、topic.prefix は topic.prefix 構成の値です。

詳細については、Debezium ドキュメントの トピック名をご覧ください。

次のステップ