建立 PostgreSQL 適用的 Cloud SQL 來源連接器

本文說明如何為 Kafka Connect 建立 PostgreSQL 適用的 Cloud SQL 來源連接器。

PostgreSQL 適用的 Cloud SQL 來源連接器是 Debezium PostgreSQL 連接器的執行個體。這項服務會從 PostgreSQL 適用的 Cloud SQL 資料庫讀取資料列層級的變更，並將這些變更寫入 Managed Service for Apache Kafka 叢集的主題。

這個連結器的用途包括：

  • 即時監控資料庫的列層級變更。
  • 將資料庫變更事件整合至事件導向架構。
  • 回應資料庫事件，例如插入或刪除資料列。
  • 將資料庫變更複製到其他系統。

事前準備

建立 PostgreSQL 適用的 Cloud SQL 來源連接器前，請確認您已備妥下列項目：

必要角色和權限

如要取得建立連結器所需的權限，請要求管理員授予您專案的「Managed Kafka Connector 編輯者」 (roles/managedkafka.connectorEditor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。

這個預先定義的角色具備建立連接器所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：

所需權限

如要建立連接器，必須具備下列權限：

  • 建立連接器： managedkafka.connectors.create

您或許還可透過自訂角色或其他預先定義的角色取得這些權限。

授予從 Cloud SQL 讀取資料的權限

代管 Kafka 服務帳戶必須具備存取 PostgreSQL 適用的 Cloud SQL 的權限。將下列 IAM 角色授予服務帳戶：

  • Cloud SQL 用戶端 (roles/cloudsql.client)
  • Cloud SQL 執行個體使用者 (roles/cloudsql.instanceUser)

代管 Kafka 服務帳戶的格式如下： service-PROJECT_NUMBER@gcp-sa-managedkafka.， 其中 PROJECT_NUMBER 是 Connect 叢集的專案編號。

如果 Connect 叢集與 Managed Service for Apache Kafka 叢集位於不同專案，請參閱「 在不同專案中建立 Connect 叢集」。

設定資料庫

建立連接器前，請務必設定資料庫複製功能，並啟用連接器，以便向資料庫驗證。下列各節將說明這些步驟。

啟用邏輯解碼

PostgreSQL 適用的 Cloud SQL 來源連接器會使用 PostgreSQL 的邏輯解碼功能。如要在 PostgreSQL 適用的 Cloud SQL 執行個體中啟用邏輯解碼功能，請按照下列步驟操作。

控制台

  1. 前往「Cloud SQL」>「執行個體」。

    前往「Instances」(執行個體)

  2. 按一下執行個體的名稱。

  3. 按一下「Edit」(編輯)。

  4. 展開「旗標和參數」。

  5. 按一下「新增資料庫旗標」。

  6. 在「Choose a flag」(選擇標記) 清單中，選取「cloudsql.logical_decoding」。

  7. 在「值」部分選取 On。

  8. 按一下「完成」。

  9. 按一下 [儲存]。

詳情請參閱「設定邏輯複製和解碼」。

設定變更資料擷取 (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 服務帳戶建立使用者帳戶

PostgreSQL 適用的 Cloud SQL 來源連接器會使用 IAM 資料庫驗證機制連線至資料庫。如要啟用 IAM 資料庫驗證，請將 Managed Kafka 服務帳戶新增至 Cloud SQL 執行個體，方法如下：

控制台

  1. 前往「Cloud SQL」>「執行個體」

    前往「Instances」(執行個體)

  2. 按一下執行個體的名稱。

  3. 在導覽窗格中，按一下「使用者」。

  4. 按一下「Add user account」(新增使用者帳戶)。

  5. 在「新增使用者帳戶」窗格中，選取「Cloud IAM」。

  6. 在「IAM principal」(IAM 主體) 欄位中輸入下列內容：

    service-PROJECT_NUMBER@gcp-sa-managedkafka.

    將 PROJECT_NUMBER 替換為 Connect 叢集的專案編號。

  7. 按一下「新增」。

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：PostgreSQL 適用的 Cloud SQL 執行個體名稱。

由於資料庫使用者名稱的長度限制，使用者名稱會捨棄 . 後置字元，因此使用者名稱為 service-PROJECT_NUMBER@gcp-sa-managedkafka.iam。執行參照 IAM 使用者帳戶的 SQL 查詢時，請指定截斷的名稱。

設定使用者帳戶

建立 IAM 使用者帳戶後，請以具備 cloudsqlsuperuser 角色的使用者 (例如預設 postgres 使用者) 身分連線至資料庫，然後執行下列 SQL 查詢。

控制台

  1. 允許使用者讀取預寫記錄。

    ALTER USER "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam" WITH REPLICATION;

  2. 授予使用者資料表的 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";

  3. 為每個資料表授予使用者存取資料表結構定義的權限。如果資料表位於預設 public 結構定義中，則可略過這個步驟。

    GRANT USAGE ON SCHEMA SCHEMA_NAME
TO "service-PROJECT_NUMBER@gcp-sa-managedkafka.iam";

設定網路

PostgreSQL 適用的 Cloud SQL 來源連接器可透過下列方式連線至 Cloud SQL 執行個體：

  • 私人 IP
  • Private Service Connect
  • 公開 IP

如要進一步瞭解這些選項，請參閱「選擇連線至 Cloud SQL 的方式」。為了安全起見，建議使用私人 IP 或 Private Service Connect，因為這些選項不需要連線至外部 IP 位址。

下表列出各選項的網路需求：

IP 位址類型 需求條件
私人 IP 為執行個體設定私人 IP。詳情請參閱「設定私人 IP」一文。
Private Service Connect
  1. 為執行個體設定 Private Service Connect，並取得 Private Service Connect 端點的 DNS 名稱。詳情請參閱「使用 Private Service Connect 連線至執行個體」。
  2. 將端點的 DNS 名稱新增至 Connect 叢集的「可解析的 DNS 網域」。詳情請參閱「 更新 Connect 叢集」。
公開 IP
  1. 為執行個體設定公開 IP。詳情請參閱「設定公開 IP」。
  2. 設定 Public NAT，讓 Connect 叢集工作人員與網際網路通訊。詳情請參閱「設定 Public NAT」。建立 Cloud NAT 閘道時，請指定包含 Connect 叢集 主要子網路的虛擬私有雲網路。

建立 PostgreSQL 適用的 Cloud SQL 來源連接器

如要建立 PostgreSQL 適用的 Cloud SQL 來源連接器，請完成下列步驟。

連接器初始化時，會執行下列動作：

  1. 建立資料庫的初始快照。
  2. 為含有資料列的每個資料表建立 Kafka 主題。
  3. 針對每個資料庫資料列，將變更事件傳送至對應的主題。

連接器執行期間，會持續將變更事件傳送至主題。如要進一步瞭解初始快照，請參閱 Debezium 說明文件中的「快照」一節。

控制台

  1. 前往 Google Cloud 控制台的「Connect Clusters」(連結叢集) 頁面。

    前往「Connect Clusters」(連結叢集)

  2. 按一下要建立連接器的 Connect 叢集。

  3. 按一下「Create connector」(建立連接器)。

  4. 輸入連接器名稱字串。

    如要查看連線器命名準則，請參閱 Managed Service for Apache Kafka 資源命名指南。

  5. 在「連接器外掛程式」部分，選取「PostgreSQL 適用的 Cloud SQL 來源」。

  6. 在「執行個體」清單中，選取 Cloud SQL 執行個體。

  7. 在「資料庫」清單中，選取 Cloud SQL 資料庫。

  8. 在「Topic prefix」(主題前置字元) 欄位中，輸入要用於 Kafka 主題名稱的前置字元。為每個 PostgreSQL 適用的 Cloud SQL 來源連接器選擇專屬前置字元。

  9. 選用步驟：在「Table names」(資料表名稱) 欄位，輸入要從中讀取變更資料的資料表清單 (以半形逗號分隔)，格式為 "schema_name"."table_name"。 如果將這個欄位留空，連接器會從資料庫中所有非系統表格讀取變更資料。

  10. 選用：在「設定」方塊中，新增設定屬性或編輯預設屬性。詳情請參閱「設定連接器」。

    您可能需要覆寫下列屬性的預設值：

    • driver.ipTypes：這個屬性必須與 Cloud SQL 執行個體的網路設定相符。請參閱「IP 位址類型」。

    • slot.name：如果為相同資料庫建立多個連接器例項，請為每個連接器指定不重複的值。請參閱「複寫位置」。

  11. 選用：選取「任務重新啟動政策」。詳情請參閱「工作重新啟動政策」。

  12. 點選「建立」。

gcloud

  1. 在 Google Cloud 控制台中啟用 Cloud Shell。

    啟用 Cloud Shell

    控制台底部會開啟 Cloud Shell 工作階段，並顯示指令列提示。 Google Cloud 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 適用的 Cloud SQL 來源連接器設定檔範例：

    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：私人 IP
  • PSC：Private Service Connect
  • PUBLIC：公開 IP

driver.ipTypes 屬性包含以半形逗號分隔的 IP 類型清單，並依偏好順序排列，例如 driver.ipTypes=PRIVATE,PUBLIC。

詳情請參閱「設定網路」。

發布作業名稱

根據預設，連接器會嘗試從名為 dbz_publication 的發布項目串流。如要指定其他發布項目，請在設定中加入 publication.name=PUBLICATION_NAME，其中 PUBLICATION_NAME 是發布項目名稱。例如：publication.name=my_publication。

複製運算單元

PostgreSQL 會使用複製位置串流資料庫資料表變更。根據預設，連接器會建立名為 debezium 的複製運算單元。如要使用其他 slot 名稱，請設定 slot.name 屬性。

如果為同一個資料庫建立兩個連接器執行個體，則必須為每個連接器指定專屬的時段名稱。

根據預設，連接器會將 slot.drop.on.stop 屬性設為 false，避免資料遺失。永久刪除連接器時，您必須手動捨棄連接器使用的複寫位置。除非使用 slot.name 屬性設定其他名稱，否則複製運算單元名稱預設為 debezium。

建議您設定快訊，監控來源 PostgreSQL 資料庫伺服器上的 WAL 磁碟用量，並捨棄任何未使用的複寫時段。

表格篩選器

根據預設，連接器會擷取資料庫中每個非系統資料表的變更資料。如要篩選要擷取的資料表，請指定下列一或多個設定：

  • 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 說明文件中的 主題名稱。

後續步驟