Cloud SQL for PostgreSQL ソースコネクタを使用すると、Cloud SQL for PostgreSQL データベースから行レベルの変更をキャプチャして、Managed Service for Apache Kafka クラスタのトピックに書き込むことができます。
このコネクタの一般的なユースケースは次のとおりです。
- 行レベルのデータベースの変更をリアルタイムでモニタリングします。
- データベース変更イベントをイベント ドリブン アーキテクチャに統合する。
- 行の挿入や削除などのデータベース イベントに応答する。
- データベースの変更を他のシステムにコピーする。
始める前に
Cloud SQL for PostgreSQL ソースコネクタを作成する前に、次のものがあることを確認してください。
データベースを含む Cloud SQL for PostgreSQL インスタンス。これらのリソースを作成する方法については、 Google Cloud コンソールを使用して Cloud SQL for PostgreSQL データベースを作成しクエリを実行するをご覧ください。
必要なロールと権限
コネクタの作成に必要な権限を取得するには、プロジェクトに対するマネージド Kafka コネクタ編集者 （roles/managedkafka.connectorEditor）IAM ロールを付与するよう管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。
この事前定義ロールには、コネクタの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。
必要な権限
コネクタを作成するには、次の権限が必要です。
-
コネクタを作成します。
managedkafka.connectors.create
カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。
Cloud SQL から読み取る権限を付与する
マネージド Kafka サービス アカウントには、Cloud SQL for PostgreSQL にアクセスする権限が必要です。サービス アカウントに次の IAM ロールを付与します。
- Cloud SQL クライアント（
roles/cloudsql.client） - Cloud SQL インスタンス ユーザー（
roles/cloudsql.instanceUser）
Managed Kafka サービス アカウントの形式は service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.com です。ここで、PROJECT_NUMBER は Connect クラスタのプロジェクト番号です。
Connect クラスタが Managed Service for Apache Kafka クラスタとは異なるプロジェクトにある場合は、 別のプロジェクトに Connect クラスタを作成するをご覧ください。
データベースを構成する
コネクタを作成する前に、データベース レプリケーションを構成し、コネクタがデータベースで認証できるようにする必要があります。以降のセクションでは、これらの手順について説明します。
論理デコードを有効にする
Cloud SQL for PostgreSQL ソースコネクタは、PostgreSQL の論理デコーディング機能に依存しています。Cloud SQL for PostgreSQL インスタンスで論理デコーディングを有効にするには、次の操作を行います。
コンソール
[Cloud SQL] > [インスタンス] に移動します。
インスタンスの名前をクリックします。
[ 編集] をクリックします。
[フラグとパラメータ] を開きます。
[データベース フラグを追加] をクリックします。
[フラグを選択] リストで、
cloudsql.logical_decodingを選択します。[値] で [
On] を選択します。[完了] をクリックします。
[保存] をクリックします。
詳細については、論理レプリケーションとデコードを設定するをご覧ください。
変更データ キャプチャ（CDC）を構成する
インスタンスで論理デコードを有効にしたら、複製するテーブルの変更データ キャプチャ（CDC）を有効にします。
テーブルで CDC を有効にするには、CREATE PUBLICATION SQL ステートメントを実行します。このステートメントは、複製するテーブルのグループを定義するパブリケーションを作成します。
オプション 1: データベース内のすべてのテーブルの変更を複製するパブリケーションを作成します。
CREATE PUBLICATION dbz_publication FOR ALL TABLES;オプション 2: 特定のテーブルセットのパブリケーションを作成します。
CREATE PUBLICATION dbz_publication FOR TABLE TABLE_LIST;TABLE_LISTは、テーブルのカンマ区切りのリストに置き換えます。形式は"schema_name"."table_name"です。図のように、スキーマ名とテーブル名を二重引用符で囲むと、名前に特殊文字や大文字が含まれている場合に構文エラーが発生するのを防ぐことができます。
デフォルトでは、コネクタはパブリケーション名に dbz_publication を使用します。別の名前のパブリケーションを使用するには、パブリケーション名をご覧ください。
Managed Kafka サービス アカウントのユーザー アカウントを作成する
Cloud SQL for PostgreSQL ソースコネクタは、IAM データベース認証を使用してデータベースに接続します。IAM データベース認証を有効にするには、次のように Managed Kafka サービス アカウントを Cloud SQL インスタンスに追加します。
コンソール
[Cloud SQL] > [インスタンス] に移動します。
インスタンスの名前をクリックします。
ナビゲーション パネルで [ユーザー] をクリックします。
[ユーザー アカウントを追加] をクリックします。
[ユーザー アカウントを追加] ペインで、[Cloud IAM] を選択します。
[IAM プリンシパル] フィールドに、次のように入力します。
service-PROJECT_NUMBER@gcp-sa-managedkafka.iam.gserviceaccount.comPROJECT_NUMBERは、Connect クラスタのプロジェクト番号に置き換えます。[追加] をクリックします。
gcloud
gcloud sql users create コマンドを実行します。
gcloud sql users create service-PROJECT_NUMBER@gcp-sa-managedkafka.iam \
--instance=INSTANCE_NAME \
--type=cloud_iam_service_account
次のように置き換えます。
PROJECT_NUMBER: Connect クラスタのプロジェクト番号。INSTANCE_NAME: Cloud SQL for PostgreSQL インスタンスの名前。
データベースのユーザー名には長さの上限があるため、ユーザー名から .gserviceaccount.com サフィックスが削除され、ユーザー名は service-PROJECT_NUMBER@gcp-sa-managedkafka.iam になります。IAM ユーザー アカウントを参照する SQL クエリを実行する場合は、切り捨てられた名前を指定します。
ユーザー アカウントを構成する
IAM ユーザー アカウントを作成したら、cloudsqlsuperuser ロールを持つユーザー（デフォルトの postgres ユーザーなど）としてデータベースに接続し、次の SQL クエリを実行します。
コンソール
ユーザーが write-ahead log を読み取れるようにします。
ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;ユーザーにテーブルに対する
SELECT権限を付与します。GRANT SELECT ON ALL TABLES IN SCHEMA "SCHEMA_NAME" TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";または、個々のテーブルに
SELECT権限を付与することもできます。このオプションを選択する場合は、コネクタのtable.include.list構成プロパティを許可されたテーブルのリストに設定する必要があります。次の SQL クエリは、単一のテーブルにSELECT権限を付与します。GRANT SELECT ON TABLE SCHEMA_NAME.TABLE_NAME TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";テーブルごとに、テーブルのスキーマへのアクセス権をユーザーに付与します。テーブルがデフォルトの
publicスキーマにある場合は、この手順を省略できます。GRANT USAGE ON SCHEMA SCHEMA_NAME TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";
ネットワークを構成する
Cloud SQL for PostgreSQL ソースコネクタは、次の方法で Cloud SQL インスタンスに接続できます。
- プライベート IP
- Private Service Connect
- パブリック IP
これらのオプションの詳細については、Cloud SQL への接続方法を選択するをご覧ください。セキュリティのベスト プラクティスとして、プライベート IP または Private Service Connect を使用することをおすすめします。これらのオプションでは、外部 IP アドレスに接続する必要がないためです。
次の表に、各オプションのネットワーク要件を示します。
| IP アドレスの種類 | 要件 |
|---|---|
| プライベート IP | インスタンスにプライベート IP を構成します。詳細については、プライベート IP を構成するをご覧ください。 |
| Private Service Connect |
|
| パブリック IP |
|
Cloud SQL for PostgreSQL ソースコネクタを作成する
Cloud SQL for PostgreSQL ソースコネクタを作成する手順は次のとおりです。
コネクタが初期化されると、次のアクションが実行されます。
- データベースの初期スナップショットを作成します。
- 行を含むテーブルごとに Kafka トピックを作成します。
- データベースの行ごとに、対応するトピックに変更イベントを送信します。
コネクタは、実行中に変更イベントをトピックに送信し続けます。初期スナップショットの詳細については、Debezium ドキュメントのスナップショットをご覧ください。
コンソール
Google Cloud コンソールで、[クラスタを接続] ページに移動します。
コネクタを作成する Connect クラスタをクリックします。
[コネクタを作成] をクリックします。
コネクタ名に文字列を入力します。
コネクタの命名方法のガイドラインについては、Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。
[コネクタ プラグイン] で、[Cloud SQL for PostgreSQL Source] を選択します。
[インスタンス] リストで、Cloud SQL インスタンスを選択します。
[データベース] リストで、Cloud SQL データベースを選択します。
[トピックの接頭辞] フィールドに、Kafka トピック名に使用する接頭辞を入力します。Cloud SQL for PostgreSQL ソースコネクタごとに一意の接頭辞を選択します。
省略可: [テーブル名] フィールドに、変更データを読み取るテーブルのカンマ区切りのリストを
"schema_name"."table_name"の形式で入力します。このフィールドを空のままにすると、コネクタはデータベース内のすべての非システム テーブルから変更データを読み取ります。省略可: [構成] ボックスで、構成プロパティを追加するか、デフォルトのプロパティを編集します。詳細については、コネクタを構成するをご覧ください。
次のプロパティのデフォルトをオーバーライドする必要がある場合があります。
driver.ipTypes: このプロパティは、Cloud SQL インスタンスのネットワーク構成と一致する必要があります。IP アドレスの種類をご覧ください。slot.name: 同じデータベースに複数のコネクタ インスタンスを作成する場合は、コネクタごとに一意の値を指定します。レプリケーション スロットをご覧ください。
省略可: [タスクの再起動ポリシー] を選択します。詳細については、タスクの再起動ポリシーをご覧ください。
[作成] をクリックします。
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 構成ファイルのパス。
Cloud SQL for PostgreSQL ソースコネクタの構成ファイルの例を次に示します。
connector.class: io.debezium.connector.postgresql.PostgresConnector database.dbname: DATABASE_NAME driver.cloudSqlInstance: INSTANCE_ID driver.enableIamAuth: "true" driver.ipTypes: IP_TYPES driver.sslmode: disable key.converter: org.apache.kafka.connect.json.JsonConverter key.converter.schemas.enable: "false" plugin.name: pgoutput slot.name: SLOT_NAME table.include.list: TABLE_LIST topic.prefix: TOPIC_PREFIX value.converter: org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable: "true"次のように置き換えます。
INSTANCE_ID: データベースを含む Cloud SQL インスタンスの ID。次の形式で指定します。PROJECT_ID:REGION:INSTANCE_NAME
DATABASE_NAME: 読み取り元の Cloud SQL データベースの名前。IP_TYPES: IP アドレスタイプのカンマ区切りのリストSLOT_NAME: 作成するレプリケーション スロットの名前。TABLE_LIST: 変更データの読み取り元のテーブルのカンマ区切りのリスト。"schema_name"."table_name"の形式で指定します。TOPIC_PREFIX: Kafka トピック名に使用する接頭辞。
コネクタを構成する
このセクションでは、コネクタに設定できる構成プロパティについて説明します。完全なリストについては、Debezium ドキュメントの PostgreSQL 用 Debezium コネクタをご覧ください。
IP アドレスの種類
driver.ipTypes プロパティは、コネクタがデータベースへの接続に使用する IP アドレスのタイプを指定します。
PRIVATE: プライベート IPPSC: Private Service ConnectPUBLIC: パブリック IP
driver.ipTypes プロパティには、優先順に IP タイプのカンマ区切りのリストが含まれます（例: driver.ipTypes=PRIVATE,PUBLIC）。
詳細については、ネットワークを構成するをご覧ください。
パブリケーション名
デフォルトでは、コネクタは 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 ドキュメントの トピック名をご覧ください。