汎用の PostgreSQL ソースコネクタを使用すると、PostgreSQL データベースから行レベルの変更をキャプチャし、Managed Service for Apache Kafka クラスタのトピックにストリーミングできます。
このコネクタの一般的なユースケースは次のとおりです。
- 行レベルのデータベースの変更をリアルタイムでモニタリングします。
- データベース変更イベントをイベント ドリブン アーキテクチャに統合する。
- 行の挿入や削除などのデータベース イベントに応答する。
- データベースの変更を他のシステムにコピーする。
- PostgreSQL テーブルを複製または復元する。
Cloud SQL for PostgreSQL データベースから変更を読み取るには、Cloud SQL for PostgreSQL ソースコネクタを使用します。
始める前に
汎用 PostgreSQL ソースコネクタを作成する前に、次のものがあることを確認してください。
PostgreSQL データベース。
Kafka クラスタに関連付けられた Connect クラスタ。
データベース パスワードを保存する Secret Manager のシークレットを作成します。構成でデータベース SSL を使用する場合は、データベース SSL パスワードのシークレットも作成します。シークレットを使用して Connect クラスタを構成します。詳細については、Secret Manager のリソースをご覧ください。
必要なロールと権限
コネクタの作成に必要な権限を取得するには、プロジェクトに対するマネージド Kafka コネクタ編集者 （roles/managedkafka.connectorEditor）IAM ロールを付与するよう管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。
この事前定義ロールには、コネクタの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。
必要な権限
コネクタを作成するには、次の権限が必要です。
-
コネクタを作成します。
managedkafka.connectors.create
カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。
Secret Manager のシークレットにアクセスする権限を付与する
マネージド 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 データベースの構成
コネクタがデータベースからデータ変更イベントを読み取れるようにするには、次の設定を構成します。
サーバーの
wal_levelをlogicalに設定します。ALTER SYSTEM SET wal_level = logical;設定を適用するには、サーバーを再起動します。
コネクタが PostgreSQL に対して認証するためのデータベース ユーザーを作成します。データベース ユーザーはレプリケーション ロールである必要があります。これにより、レプリケーション モードでサーバーに接続できます。
CREATE ROLE ROLE_NAME WITH REPLICATION LOGIN PASSWORD 'ROLE_PASSWORD';次のように置き換えます。
ROLE_NAME: ユーザーの名前（例:debezium_user）。ROLE_PASSWORD: ユーザーのパスワード。
キャプチャするテーブルのパブリケーションを作成します。コネクタは、パブリケーションに登録してデータ変更イベントを受信します。
CREATE PUBLICATION dbz_publication FOR TABLE "SCHEMA_NAME"."TABLE_NAME";次のように置き換えます。
SCHEMA_NAME: テーブルのスキーマ。TABLE_NAME: テーブル名。
名前が特殊文字や大文字を含む場合に構文エラーを回避するため、スキーマ名とテーブル名を二重引用符で囲むことをおすすめします。
または、データベース内のすべてのテーブルの変更を複製するパブリケーションを作成することもできます。
CREATE PUBLICATION dbz_publication FOR ALL TABLES;コネクタの
publication.autocreate.mode設定に応じて、パブリケーションを手動で作成するか、コネクタに自動的に作成させることができます。詳細については、パブリケーション モードをご覧ください。各テーブルについて、テーブルに対する
SELECT権限をデータベース ユーザーに付与します。GRANT SELECT ON TABLE "SCHEMA_NAME"."TABLE_NAME" TO ROLE_NAME;または、スキーマ内のすべてのテーブルに対する選択権限を付与することもできます。
GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO ROLE_NAME;各テーブルについて、テーブル スキーマに対する
USAGE権限をデータベース ユーザーに付与します。テーブルがデフォルトのpublicスキーマにある場合は、この手順をスキップできます。GRANT USAGE ON SCHEMA "SCHEMA_NAME" TO ROLE_NAME;
汎用 PostgreSQL ソースコネクタを作成する
汎用 PostgreSQL ソースコネクタを作成する手順は次のとおりです。
コネクタが初期化されると、次のアクションが実行されます。
- データベースの初期スナップショットを作成します。
- 行を含むテーブルごとに Kafka トピックを作成します。
- データベースの行ごとに、対応するトピックに変更イベントを送信します。
コネクタは、実行中に変更イベントをトピックに送信し続けます。初期スナップショットの詳細については、Debezium ドキュメントのスナップショットをご覧ください。
コンソール
Google Cloud コンソールで、[クラスタを接続] ページに移動します。
コネクタを作成する Connect クラスタをクリックします。
[コネクタを作成] をクリックします。
コネクタ名に文字列を入力します。
コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。
[コネクタ プラグイン] で、[汎用 PostgreSQL ソース] を選択します。
[データベースのホスト名] フィールドに、PostgreSQL サーバーのホスト名または IP アドレスを入力します。
[データベース名] フィールドに、データベースの名前を入力します。
[データベース ユーザー] フィールドに、レプリカロールの名前を入力します。コネクタは、このロールを使用して PostgreSQL サーバーを認証します。
[トピックの接頭辞] フィールドに、Kafka トピック名に使用する接頭辞を入力します。
[シークレット] リストで、データベース パスワードを含むシークレットを選択します。
省略可: [構成] ボックスで、構成プロパティを追加するか、デフォルトのプロパティを編集します。詳細については、コネクタを構成するをご覧ください。
省略可: [タスクの再起動ポリシー] を選択します。詳細については、タスクの再起動ポリシーをご覧ください。
[作成] をクリックします。
gcloud
-
Google Cloud コンソールで Cloud Shell をアクティブにします。
Google Cloud コンソールの下部で Cloud Shell セッションが開始し、コマンドライン プロンプトが表示されます。Cloud Shell はシェル環境です。Google Cloud CLI がすでにインストールされており、現在のプロジェクトの値もすでに設定されています。セッションが初期化されるまで数秒かかることがあります。
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 の Secret へのパス。シークレットは次の形式で指定します。
${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: プライベート IPPSC: Private Service ConnectPUBLIC: パブリック 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と併用することはできません。
トピック名
デフォルトでは、コネクタは topic_prefix.schema.table_name という命名規則で Kafka トピックを作成します。ここで、topic.prefix は topic.prefix 構成の値です。
詳細については、Debezium ドキュメントの トピック名をご覧ください。