本文說明如何建立一般 PostgreSQL 來源連接器。
一般 PostgreSQL 來源連接器是 Debezium PostgreSQL 連接器的執行個體。這個連接器會從 PostgreSQL 資料庫讀取資料列層級的變更，並將這些變更寫入 Managed Service for Apache Kafka 叢集的主題。
這個連結器的用途包括：
- 即時監控資料庫的列層級變更。
- 將資料庫變更事件整合至事件導向架構。
- 回應資料庫事件，例如插入或刪除資料列。
- 將資料庫變更複製到其他系統。
- 複製或還原 PostgreSQL 資料表。
事前準備
建立一般 PostgreSQL 來源連接器前，請確認您已備妥下列項目：
PostgreSQL 資料庫。
與 Kafka 叢集相關聯的 Connect 叢集。
建立 Secret Manager 密鑰，儲存資料庫密碼。如果設定使用資料庫 SSL，請一併建立資料庫 SSL 密碼的密鑰。使用密碼設定 Connect 叢集。詳情請參閱「Secret Manager 資源」。
必要角色和權限
如要取得建立連結器所需的權限，請要求管理員授予您專案的「Managed Kafka Connector 編輯者」 (roles/managedkafka.connectorEditor) IAM 角色。如要進一步瞭解如何授予角色，請參閱「管理專案、資料夾和組織的存取權」。
這個預先定義的角色具備建立連接器所需的權限。如要查看確切的必要權限，請展開「Required permissions」(必要權限) 部分：
所需權限
如要建立連接器，必須具備下列權限：
-
建立連接器：
managedkafka.connectors.create
授予存取 Secret Manager 密鑰的權限
代管 Kafka 服務帳戶需要權限，才能查看及存取儲存在 Secret Manager 中的密鑰。 將下列 IAM 角色授予服務帳戶：
- Secret Manager 檢視者 (
roles/secretmanager.viewer) - Secret Manager 密鑰存取者 (
roles/secretmanager.secretAccessor)
代管 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 Clusters」(連結叢集) 頁面。
按一下要建立連接器的 Connect 叢集。
按一下「Create connector」(建立連接器)。
輸入連接器名稱字串。
如要查看連線器命名準則，請參閱 Managed Service for Apache Kafka 資源命名指南。
在「連接器外掛程式」部分，選取「一般 PostgreSQL 來源」。
在「資料庫主機名稱」欄位中，輸入 PostgreSQL 伺服器的主機名稱或 IP 位址。
在「Database name」(資料庫名稱) 欄位中輸入資料庫名稱。
在「Database user」(資料庫使用者) 欄位中，輸入副本角色的名稱。連接器會使用這個角色向 PostgreSQL 伺服器進行驗證。
在「Topic prefix」(主題前置字元) 欄位中，輸入要用於 Kafka 主題名稱的前置字元。
在「Secret」清單中，選取含有資料庫密碼的 Secret。
選用：在「設定」方塊中，新增設定屬性或編輯預設屬性。詳情請參閱「設定連接器」。
選用：選取「任務重新啟動政策」。詳情請參閱「工作重新啟動政策」。
點選「建立」。
gcloud
-
在 Google Cloud 控制台中啟用 Cloud Shell。
控制台底部會開啟 Cloud Shell 工作階段，並顯示指令列提示。 Google Cloud 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 密鑰的路徑，內含資料庫密碼。請使用下列格式指定密鑰：
${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 名稱，請設定 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 說明文件中的 主題名稱。