GaussDB主备版数据同步到Kafka

场景描述

本实践通过创建DRS同步任务，实现将源数据库GaussDB主备版的增量数据同步到目标端Kafka。

前提条件

• 拥有华为云实名认证账号。
• 账户余额大于等于0元。

• 已登录数据复制服务控制台。

服务列表

• 虚拟私有云 VPC
• 云数据库GaussDB主备版

• 分布式消息服务Kafka
• 数据复制服务 DRS

GaussDB主备数据库架构说明

图1 GaussDB主备逻辑架构图

GaussDB主备版主要包含了OM、CM和DN等模块，主备版因为无CN，因此是业务应用直接下发任务给DN， DN处理完成后再将结果返回给业务应用。

主备版适用于数据量较小，且长期来看数据不会大幅度增长，但是对数据的可靠性，以及业务的可用性有一定诉求的场景。

DRS同步网络示意图

本示例中，DRS源数据库为华为云GaussDB主备版，目标端为分布式消息服务Kafka，通过VPC网络，将源数据库的增量数据同步到目标端，部署架构可参考如下图示：

图2 GaussDB主备版到Kafka同VPC场景

使用说明

• 本实践的资源规划仅作为演示，实际业务场景资源以用户实际需求为准。
• 本实践端到端的数据为测试数据，仅供参考；更多关于DRS使用相关内容请单击这里了解。

资源规划

操作流程

创建DRS任务，并且将GaussDB主备版数据增量同步到Kafka的主要任务流程如图所示。

创建VPC

创建VPC，为创建GaussDB实例准备网络资源。

1. 登录华为云控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击左侧的服务列表图标，选择“网络 > 虚拟私有云 VPC”。

   进入虚拟私有云信息页面。

4. 单击“创建虚拟私有云”购买VPC。

   

5. 单击“立即创建”。
6. 返回VPC列表，查看创建VPC是否创建完成。

   当VPC列表的VPC状态为“可用”时，表示VPC创建完成。

创建安全组

创安全组，为创建GaussDB实例准备安全组。

1. 登录华为云控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击左侧的服务列表图标，选择“网络 > 虚拟私有云 VPC”。

   进入虚拟私有云信息页面。

4. 选择“访问控制 > 安全组”。
5. 单击“创建安全组”。
6. 填写安全组名称等信息。

   

7. 单击“确定”。

创建GaussDB主备实例

本章节介绍创建GaussDB主备实例，作为同步任务目标库。

1. 登录华为云控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击左侧的服务列表图标，选择“数据库 > 云数据库 GaussDB”。
4. 单击“购买数据库实例”。
5. 配置实例名称和实例基本信息。

   

6. 选择实例规格。

   

   本示例中为测试实例，选择较小的测试规格，实际可选规格以界面为准。

7. 选择实例所属的VPC和安全组（创建VPC和创建安全组），配置数据库端口。

   

8. 配置实例密码等信息。

   

9. 单击“立即购买”，确认信息并提交。
10. 返回实例列表。

    当实例运行状态为“正常”时，表示实例创建完成。

GaussDB实例构造数据

1. 登录华为云控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击左侧的服务列表图标，选择“数据库 > 云数据库 GaussDB”。
4. 选择GaussDB实例，单击实例后的“更多 > 登录”。
5. 在弹出的对话框中输入密码后，单击“测试连接”检查。
6. 连接成功后单击“登录”，登录GaussDB实例。
7. 单击“新建数据库”，创建db_test测试库。

   

8. 在db_test库中执行如下语句，创建对应的测试表schema_test.table1。

   create table schema_test.table1(c1 int primary key,c2 varchar(10),c3 TIMESTAMP(6));

创建Kafka实例

本章节介绍创建Kafka实例。

1. 登录华为云控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击左侧的服务列表图标，选择“应用中间件 > 分布式消息服务Kafka版”。
4. 单击“购买Kafka实例”。
5. 选择实例区域和可用区。

   

6. 配置实例名称和实例规格等信息。

   

7. 选择存储空间和容量阈值策略。

   

8. 选择实例所属的VPC和安全组（创建VPC和创建安全组）。

   

9. 配置实例密码。

   

10. 单击“立即购买”，确认信息无误后单击“提交”。
11. 返回实例列表。

    当Kafka实例运行状态为“运行中”时，表示实例创建完成。

创建Kafka的Topic

1. 在“Kafka专享版”页面，单击Kafka实例的名称。
2. 选择“Topic管理”页签，单击“创建Topic”。
3. 在弹出的“创建Topic”的对话框中，填写Topic名称和配置信息，单击“确定”，完成创建Topic。

   

创建DRS同步任务

1. 登录华为云控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击左侧的服务列表图标，选择“数据库 > 数据复制服务 DRS”。
4. 选择左侧“实时同步管理”，单击“创建同步任务”。
5. 填写同步任务参数：
   1. 配置同步任务名称。

      

   2. 选择需要同步任务的源库、目标数据库以及网络信息。

      这里的源库选择创建GaussDB主备实例中创建的GaussDB实例。

      

   3. 选择规格类型和可用区。

      

6. 单击“开始创建”。

   同步实例创建中，大约需要5-10分钟。

7. 配置源库信息和目标库数据库密码。
   1. 配置源库信息。
   2. 单击“测试连接”。

      当界面显示“测试成功”时表示连接成功。

      

   3. 选择目标库所在VPC和子网，填写Kafka的IP地址和端口。
   4. 单击“测试连接”。

      当界面显示“测试成功”时表示连接成功。

      

8. 单击“下一步”。
9. 选择同步信息、策略、消息格式和对象等，投递到Kafka的消息格式。

   本次选择如下。

   

   表2 同步设置

   类别	设置
   同步Topic策略	集中投递到一个Topic，Topic名称“testTopic”。 不同Topica策略，对应选择的Kafka partition策略也不同，详细的说明可参考同步Topic和Partition策略说明。
   同步到Kafka partition策略	全部投递到同Partition 0。 不同Topica策略，对应选择的Kafka partition策略也不同，详细的说明可参考同步Topic和Partition策略说明。
   投递到Kafka的数据格式	可选择JSON格式，可参考Kafka消息格式。
   同步对象	同步对象选择db_test库下的schema_test.table1。

10. 单击“下一步”，等待预检查结果。
11. 当所有检查都是“通过”时，单击"下一步”。
12. 确认同步任务信息正确后，单击“启动任务”。

    返回DRS实时同步管理，查看同步任务状态。

    启动中状态一般需要几分钟，请耐心等待。

    当状态变更为“增量同步”，表示同步任务已启动。

    

    • 当前示例中GaussDB主备到Kafka选择单增量同步，任务启动后为增量同步状态。
    • 如果创建的任务为全量+增量同步，任务启动后先进入全量同步，全量数据同步完成后进入增量同步状态。
    • 增量同步会持续性同步增量数据，不会自动结束。

确认同步任务执行结果

由于本次实践为增量同步模式，DRS任务会将源库的产生的增量数据持续同步至目标库中，直到手动任务结束。下面我们通过在源库GaussDB中插入数据，查看Kafka的接收到的数据来验证同步结果。

1. 登录华为云控制台。
2. 单击管理控制台左上角的，选择区域。
3. 单击左侧的服务列表图标，选择“数据库 > 云数据库 GaussDB””。
4. 单击GaussDB实例后的“更多 > 登录”。
5. 在弹出的对话框中，输入实例密码，单击“测试连接”检查。
6. 连接成功后单击“登录”，登录GaussDB实例。
7. 在DRS同步对象的db_test.schema_test.table1表中，执行如下语句，插入数据。

   insert into schema_test.table1 values(1,'testKafka',current_timestamp(6)); 
   update schema_test.table1 set c2 ='G2K' where c1 =1;
   delete schema_test.table1 where  c1 =1;

8. 通过Kafka客户端查看接收到JSON格式数据。

    ./kafka-console-consumer.sh --bootstrap-server ip:port --topic  testTopic   --from-beginning

   

9. 结束同步任务。
   根据业务情况，确认数据已全部同步至目标库，可以结束当前任务。

   1. 单击“操作”列的“结束”。
   2. 仔细阅读提示后，单击“是”，结束任务。

同步Topic和Partition策略说明