使用Debezium将PostgreSQL数据同步至消息队列Kafka版

使用限制

Debezium PostgreSQL Source Connector只能配置一个Task用于消费源端的CDC数据，不支持并发消费配置。

前提条件

• 开通事件总线EventBridge并授权。

• 已开通对象存储OSS服务并创建存储空间（Bucket）。更多信息，请参见控制台创建存储空间。

• 已开通Serverless应用引擎服务。更多信息，请参见准备工作。

• 已创建专有网络及交换机。更多信息，请参见快速搭建（ROS）。

• 已购买并部署云消息队列 Kafka 版实例。更多信息，请参见购买并部署实例。

步骤一：创建数据表

1. 登录RDS管理控制台，创建RDS PostgreSQL实例。更多信息，请参见创建RDS PostgreSQL实例。

   创建实例时，请选择与前提条件中已购买部署的Kafka实例相同的VPC，并将此VPC网段加入白名单。

2. 实例创建完成后，在实例列表页面单击目标实例，然后在实例详情页面的左侧导航栏，完成以下操作。

   1. 创建一个新账号，也可使用已有账号。更多信息，请参见创建账号和数据库。

   2. 创建一个数据库，也可使用已有数据库。更多信息，请参见创建账号和数据库。

   3. 单击数据库连接，记录内网地址和端口号。

      

   4. 单击参数设置，将wal_level参数的运行参数值修改为logical，修改完成后单击提交参数。

3. 在实例详情页面，单击登录数据库进入DMS数据管理服务平台，完成以下操作。

   1. 右键单击目标数据库，选择模式管理，然后单击创建模式创建新的模式（Schema）。

      说明

      Connector配置中需使用新创建的Schema，不能用系统Schema（information、pg_catalog、public）代替。

   2. 在新建的Schema中，使用SQL语句创建表格。例如，创建一个列参数分别为id和number的表格，命令如下。更多信息，请参见SQL Commands。

      CREATE TABLE sql_table(id INT ,number INT);

   3. 执行以下命令，创建初始化wal2json插件，开启数据订阅能力。

      SELECT 'init' FROM pg_create_logical_replication_slot('test_slot', 'wal2json');

步骤二：创建Connector任务

1. 下载Debezium PostgreSQL Source Connector文件，上传至提前创建好的OSS Bucket。更多信息，请参见控制台上传文件。

   重要

   下载Debezium PostgreSQL Connector文件时请选择适配Java 8的版本。

2. 登录云消息队列 Kafka 版控制台，在概览页面的资源分布区域，选择地域。

3. 在左侧导航栏，选择Connector生态集成 > 任务列表。

4. 在任务列表页面，单击创建任务。

5. 在创建任务面板，设置任务名称，配置以下配置项。

   • 任务创建

     1. 在Source（源）配置向导，选择数据提供方为Apache Kafka Connect，单击下一步。

     2. 在连接器配置配置向导，设置以下配置项，然后单击下一步。

        配置项	参数	说明
        Kafka Connect插件	Bucket存储桶	选择OSS Bucket。
        	文件	选择上传的.ZIP文件。
        Kafka资源信息	Kafka参数配置	选择Source Connect。
        	Kafka实例	选择前提条件中创建的实例。
        	专有网络VPC	选择VPC ID。
        	交换机	选择vSwitch ID。
        	安全组	选择安全组。
        Kafka Connect配置信息	解析当前ZIP包下的properties文件	选择新建properties文件。在输入框中更新相关字段的取值。 展开查看字段描述 字段名 描述 connector.class 运行的Connector包名称，无需修改。 database.dbname 数据库名称。 database.hostname 填写步骤一：创建数据表中获取的内网地址。 database.port 填写步骤一：创建数据表中获取的端口号。 database.user 数据库登录用户名。 database.password 数据库登录密码。 slot.name 数据库逻辑复制流的名称。 table.whitelist 数据库表格列表。不同表格之间用英文逗号（,）分隔，表格的指定规则为{schemaName}.{tableName}。 database.server.name 目标Topic前缀。目标Topic的命名规范为{database.server.name}.{schemaName}.{tableName}。 重要 在投递数据前，请按照命名规范提前创建好目标Topic。 展开查看示例代码 connector.class=io.debezium.connector.postgresql.PostgresConnector database.dbname=test_database database.hostname=pgm-xxx.pg.rds.aliyuncs.com database.password=xxx database.port=5432 database.user=xxx name=debezium-psql-source # 插件名称，本文场景下使用的是wal2json。有效取值包含decoderbufs、wal2json、wal2json_rds、wal2json_streaming以及wal2json_rds_streaming。 plugin.name=wal2json slot.drop_on_stop=true slot.name=test_slot # 源数据库表格列表，不同表格之间用英文逗号（,）分隔，表格的指定规则为<schemaName>.<tableName>。 table.whitelist=test_schema.test_table # 注意，这里只能有1个task进行消费。 tasks.max=1 # 目标Topic前缀，目标Topic的命名规格为<database.server.name>.<schemaName>.<tableName>。 # 例如在本文的场景下，schemaName是kafka_connect_schema，tableName是table2_with_pk # 那么，这个表格的CDC数据会流入目标topic test-prefix.kafka_connect_schema.table2_with_pk database.server.name=test-prefix # 消息Value格式转换组件。 value.converter=org.apache.kafka.connect.json.JsonConverter # 消息 Value 内容中是否包含结构体schema信息。 value.converter.schemas.enable=false Connector全量参数，请参见Debezium Connector Properties。

     3. 在实例配置配置向导，设置以下参数，然后单击下一步。

        配置项	参数	说明
        Worker规格	Worker规格	选择合适的Worker规格。
        	最小Worker数	设置为1。
        	最大Worker数	设置为1。
        Kafka Connect Worker配置	自动创建Kafka Connect Worker依赖资源	建议勾选此项，会在选择的Kafka实例中自动创建Kafka Connect运行所需的一些Internal Topic以及ConsumerGroup，并将这些必填配置自动填入配置框中。包括以下配置项： Offset Topic：用于存储源数据偏移量，命名规则为connect-eb-offset-<任务名称>。 Config Topic：用于存储Connectors以及Tasks的配置信息，命名规则为connect-eb-config-<任务名称>。 Status Topic：用于存储Connectors以及Tasks状态信息，命名规则为connect-eb-status-<任务名称>。 Kafka Connect Consumer Group：Kafka Connect Worker用于消费Internal Topics的消费组，命名规则为connect-eb-cluster-<任务名称>。 Kafka Source Connector Consumer Group：只针对Sink Connector有效，用于消费源Kafka Topic中的数据，命名规则为connector-eb-cluster-<任务名称>-<connector名称>。

     4. 在运行配置配置向导，将日志投递方式设置为投递至SLS或者投递至Kafka，在角色授权卡片设置Connect依赖的角色配置，然后单击保存。

        重要

        建议配置的角色包含AliyunSAEFullAccess权限，否则可能会导致任务运行失败。

   • 任务属性

     设置此任务的重试策略及死信队列。更多信息，请参见重试和死信。

   等待任务状态变为运行中，此时Connector已经在正常工作中。

步骤三：测试Connector任务

1. 在DMS数据管理服务平台，向步骤一：创建数据表中创建的数据表插入一条数据。例如，插入一条id为123，number为20000的数据，命令如下。

   INSERT INTO sql_table(id, number) VALUES(123,20000);

2. 登录云消息队列 Kafka 版控制台，在实例列表页面，单击目标实例。

3. 在目标实例页面，单击目标Topic（投递数据前提前创建好的命名规则为{database.server.name}.{schemaName}.{tableName}的Topic），然后单击消息查询，查看插入的消息数据，消息Value示例如下。

   {"before":null,"after":{"id":123,"number":20000},"source":{"version":"0.9.2.Final","connector":"postgresql","name":"test-prefix","db":"wb","ts_usec":168386295815075****,"txId":10339,"lsn":412719****,"schema":"test_schema","table":"sql_table","snapshot":false,"last_snapshot_record":null},"op":"c","ts_ms":168386295****}

常见报错

场景一：所有Tasks运行失败

All tasks under connector mongo-source failed, please check the error trace of the task.

解决方法：在消息流入任务详情页面，单击基础信息区域的诊断链接，即可跳转到Connector监控页面，可以看到Tasks运行失败的详细错误信息。

场景二：Kafka Connect退出

Kafka connect exited! Please check the error log /opt/kafka/logs/connect.log on sae application to find out the reason why kafka connect exited and update the event streaming with valid arguments to solve it.

基于错误信息提示进行修复操作后，可以重新启动对应任务。

场景三：Connector参数校验失败

Start or update connector xxx failed. Error code=400. Error message=Connector configuration is invalid and contains the following 1 error(s):
Value must be one of never, initial_only, when_needed, initial, schema_only, schema_only_recovery
You can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`

curl -i -X PUT -H "Accept:application/json" -H  "Content-Type:application/json" -d @$CONNECTOR_PROPERTIES_MAPPING http://localhost:8083/connector-plugins/io.confluent.connect.jdbc.JdbcSinkConnector/config/validate

"value":{
    "name":"snapshot.mode",
    "value":null,
    "recommended_values":[
        "never",
        "initial_only",
        "when_needed",
        "initial",
        "schema_only",
        "schema_only_recovery"
    ],
    "errors":[
        "Value must be one of never, initial_only, when_needed, initial, schema_only, schema_only_recovery"
    ],
    "visible":true
}