このドキュメントでは、Kafka Connect 用の Cloud SQL for PostgreSQL ソースコネクタを作成する方法について説明します。
Cloud SQL for PostgreSQL ソースコネクタは、 Debezium PostgreSQL コネクタのインスタンスです。 Cloud SQL for PostgreSQL データベースから行レベルの変更を読み取り、 Managed Service for Apache Kafka クラスタのトピックに書き込みます。
このコネクタのユースケースは次のとおりです。
- 行レベルのデータベースの変更をリアルタイムでモニタリングする。
- データベースの変更イベントをイベント ドリブン アーキテクチャに統合する。
- 行の挿入や削除などのデータベース イベントに対応する。
- データベースの変更を他のシステムにコピーする。
始める前に
Cloud SQL for PostgreSQL ソースコネクタを作成する前に、次のものがあることを確認してください。
データベースを含む Cloud SQL for PostgreSQL インスタンス。これらのリソースを作成する方法については、 コンソールを使用して Cloud SQL for PostgreSQL データベースを作成し、クエリを実行する Google Cloud をご覧ください。
必要なロールと権限
コネクタの作成に必要な権限を取得するには、プロジェクトに対するManaged Kafka コネクタ編集者 （roles/managedkafka.connectorEditor）IAM ロールの付与を管理者に依頼してください。ロールの付与については、プロジェクト、フォルダ、組織に対するアクセス権の管理をご覧ください。
この事前定義ロールには コネクタの作成に必要な権限が含まれています。必要とされる正確な権限については、「必要な権限」セクションを開いてご確認ください。
必要な権限
コネクタを作成するには、次の権限が必要です。
-
コネクタを作成する:
managedkafka.connectors.create
カスタムロールや他の事前定義ロールを使用して、これらの権限を取得することもできます。
Cloud SQL から読み取る権限を付与する
Managed 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] [>] [インスタンス] に移動します。
インスタンスの名前をクリックします。
[ Edit] をクリックします。
[フラグとパラメータ] を開きます。
[データベース フラグを追加] をクリックします。
[フラグの選択] リストで、
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 ログを読み取れるようにします。
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 クラスタ] ページに移動します。
コネクタを作成する Connect クラスタをクリックします。
[コネクタを作成] をクリックします。
コネクタ名に文字列を入力します。
コネクタの命名方法のガイドラインについては、 Managed Service for Apache Kafka リソースの命名ガイドラインをご覧ください。
[コネクタ プラグイン] で [Cloud SQL for PostgreSQL ソース] を選択します。
[インスタンス] リストで、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 ドキュメントの Debezium コネクタ for PostgreSQL をご覧ください。
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 に設定します。コネクタを完全に削除する場合は、コネクタが使用していたレプリケーション スロットを手動で削除する必要があります。レプリケーション スロット名は、
別の構成にしない限り、デフォルトで debeziumになります。slot.name
ソース 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 ドキュメントの トピック名をご覧ください。