借助通用 PostgreSQL 源连接器，您可以从 PostgreSQL 数据库捕获行级更改，并将其流式传输到 Managed Service for Apache Kafka 集群中的主题。
此连接器的典型应用场景包括：
- 实时监控行级数据库变更。
- 将数据库变更事件集成到事件驱动型架构中。
- 响应数据库事件，例如行插入或删除。
- 将数据库更改复制到其他系统。
- 复制或恢复 PostgreSQL 表。
如需从 Cloud SQL for PostgreSQL 数据库读取更改，请使用 Cloud SQL for PostgreSQL Source 连接器。
准备工作
在创建通用 PostgreSQL 源连接器之前，请确保您具备以下条件：
PostgreSQL 数据库。
与 Kafka 集群关联的 Connect 集群。
创建用于存储数据库密码的 Secret Manager Secret。如果您的配置使用数据库 SSL，请同时为数据库 SSL 密码创建 Secret。使用 Secret 配置 Connect 集群。如需了解详情，请参阅 Secret Manager 资源。
所需的角色和权限
如需获得创建连接器所需的权限，请让您的管理员为您授予项目的 Managed Kafka Connector Editor (roles/managedkafka.connectorEditor) IAM 角色。如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。
此预定义角色可提供创建连接器所需的权限。如需查看所需的确切权限，请展开所需权限部分：
所需权限
创建连接器需要以下权限：
-
创建连接器：
managedkafka.connectors.create
授予访问 Secret Manager 密钥的权限
Managed Kafka 服务账号需要有权查看和访问存储在 Secret Manager 中的 Secret。 向服务账号授予以下 IAM 角色：
- Secret Manager Viewer (
roles/secretmanager.viewer) - Secret Manager Secret Accessor (
roles/secretmanager.secretAccessor)
Managed Kafka 服务账号采用以下格式：service-PROJECT_NUMBER@gcp-sa-managedkafka.，其中 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;或者，您也可以授予对架构中所有表的 SELECT 权限：
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 主题名称的前缀。
在Secret列表中，选择包含数据库密码的 Secret。
可选：在配置框中，添加配置属性或修改默认属性。如需了解详情，请参阅配置连接器。
可选：选择任务重启政策。如需了解详情，请参阅任务重启政策。
点击创建。
gcloud
-
在 Google Cloud 控制台中，激活 Cloud Shell。
Cloud Shell 会话随即会在 Google Cloud 控制台的底部启动，并显示命令行提示符。Cloud Shell 是一个已安装 Google Cloud CLI 且已为当前项目设置值的 Shell 环境。该会话可能需要几秒钟时间来完成初始化。
运行
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 配置中仅支持 Secret 路径。后端要求这些配置使用以下格式：${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 属性。
如果您为同一数据库创建两个连接器实例，则必须为每个连接器指定唯一的槽名称。
默认情况下，连接器会将
slot.drop.on.stop 属性设置为 false，以防止数据丢失。永久删除连接器时，您必须手动删除连接器使用的复制槽。复制槽名称默认为 debezium，除非使用 slot.name 属性进行了不同的配置。
我们建议您设置提醒，以监控源 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 文档中的 主题名称。