# GaussDB集中式版数据同步到Kafka #### 场景描述 本实践通过创建DRS同步任务，实现将源数据库GaussDB集中式版的增量数据同步到目标端Kafka。 #### 前提条件 - 拥有华为云实名认证账号。 - 账户余额大于等于0元。 - 已登录数据复制服务控制台。 #### 服务列表 - 虚拟私有云 VPC - 云数据库GaussDB集中式版 - 分布式消息服务Kafka - 数据复制服务 DRS #### GaussDB主备数据库架构说明 图1GaussDB主备逻辑架构图 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930520214.png) GaussDB集中式版主要包含了OM、CM和DN等模块，主备版因为无CN，因此是业务应用直接下发任务给DN， DN处理完成后再将结果返回给业务应用。 主备版适用于数据量较小，且长期来看数据不会大幅度增长，但是对数据的可靠性，以及业务的可用性有一定诉求的场景。 #### DRS同步网络示意图 本示例中，DRS源数据库为华为云GaussDB集中式版，目标端为分布式消息服务Kafka，通过VPC网络，将源数据库的增量数据同步到目标端，部署架构可参考如下图示： 图2GaussDB集中式版到Kafka同VPC场景 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930685842.png) #### 使用说明 - 本实践的资源规划仅作为演示，实际业务场景资源以用户实际需求为准。 - 本实践端到端的数据为测试数据，仅供参考；更多关于DRS使用相关内容请单击[这里](https://support.huaweicloud.com/realtimesyn-drs/drs_04_0447.html)了解。 #### 资源规划 表1资源规划 | 类别 | 子类 | 规划 | 备注 | |:---|:---|:---|:---| | VPC | VPC名称 | vpc-DRStest | 自定义，易理解可识别。 | | VPC | 子网网段 | 10.0.0.0/24 | 子网选择时建议预留足够网络资源 | | VPC | 所属Region | 华南-广州 | 选择和自己业务区最近的Region，减少网络时延。 | | VPC | 子网名称 | subnet-drs01 | 自定义，易理解可识别。 | | GaussDB（源库） | 实例名 | drs-gaussdb-src-1 | 自定义，易理解可识别。 | | GaussDB（源库） | 数据库版本 | GaussDB 8.1.0 | - | | GaussDB（源库） | 实例类型 | 主备版 | 参考GaussDB数据库类型说明，选择适合自己业务的库类型。 | | GaussDB（源库） | 存储 | 超高IO | GaussDB支持"超高IO"存储类型，最大吞吐量为350MB/S。 | | GaussDB（源库） | 规格 | 通用型 4 vCPUs \| 16 GB | 根据自己业务承载选择规格。 | | Kafka（目标端） | Kafka实例名 | kafka-drs | 自定义，易理解可识别。 | | Kafka（目标端） | 版本 | 2.3.0 | - | | Kafka（目标端） | 可用区 | 可用区三 | 可选择1个或者3个及以上可用区。实际业务场景推荐选择创建在不同的可用区，提升业务可靠性。 | | Kafka（目标端） | 规格 | c6.2u4g.cluster | - | | Kafka（目标端） | 代理个数 | 3 | - | | Kafka（目标端） | 存储空间 | 高I/O，200GB | 存储空间主要用于存储消息（包含副本，Kafka默认使用3副本），除了存储消息外还需要预留部分空间用于存储日志和元数据。 | | DRS同步任务 | 同步任务名 | DRS-GaussDBToKafka | 自定义。 | | DRS同步任务 | 源数据库引擎 | GaussDB集中式版 | - | | DRS同步任务 | 目标数据库引擎 | Kafka | 本示例中目标数据库为Kafka | | DRS同步任务 | 网络类型 | VPC网络 | 创建任务的时候选择"VPN、专线网络"。 | #### 操作流程 创建DRS任务，并且将GaussDB集中式版数据增量同步到Kafka的主要任务流程如图所示。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930700952.png) #### 创建VPC 创建VPC，为创建GaussDB实例准备网络资源。 1. 登录[华为云控制台](https://console.huaweicloud.com)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957458901.png)，选择区域。 3. 单击左侧的服务列表图标，选择"网络 \> 虚拟私有云 VPC"。 进入虚拟私有云信息页面。 4. 单击"创建虚拟私有云"购买VPC。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930251392.png) 5. 单击"立即创建"。 6. 返回VPC列表，查看创建VPC是否创建完成。 当VPC列表的VPC状态为"可用"时，表示VPC创建完成。 #### 创建安全组 创安全组，为创建GaussDB实例准备安全组。 1. 登录[华为云控制台](https://console.huaweicloud.com)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957538701.png)，选择区域。 3. 单击左侧的服务列表图标，选择"网络 \> 虚拟私有云 VPC"。 进入虚拟私有云信息页面。 4. 选择"访问控制 \> 安全组"。 5. 单击"创建安全组"。 6. 填写安全组名称等信息。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930419474.png) 7. 单击"确定"。 #### 创建GaussDB主备实例 本章节介绍创建GaussDB主备实例，作为同步任务目标库。 1. 登录[华为云控制台](https://console.huaweicloud.com)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930527746.png)，选择区域。 3. 单击左侧的服务列表图标，选择"数据库 \> 云数据库 GaussDB"。 4. 单击"购买数据库实例"。 5. 配置实例名称和实例基本信息。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000002109168338.png) 6. 选择实例规格。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930545232.png) 本示例中为测试实例，选择较小的测试规格，实际可选规格以界面为准。 7. 选择实例所属的VPC和安全组（[创建VPC]和[创建安全组]），配置数据库端口。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957646381.png) 8. 配置实例密码等信息。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930687130.png) 9. 单击"立即购买"，确认信息并提交。 10. 返回实例列表。 当实例运行状态为"正常"时，表示实例创建完成。 #### GaussDB实例构造数据 1. 登录[华为云控制台](https://console.huaweicloud.com)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957646385.png)，选择区域。 3. 单击左侧的服务列表图标，选择"数据库 \> 云数据库 GaussDB"。 4. 选择GaussDB实例，单击实例后的"更多 \> 登录"。 5. 在弹出的对话框中输入密码后，单击"测试连接"检查。 6. 连接成功后单击"登录"，登录GaussDB实例。 7. 单击"新建数据库"，创建db_test测试库。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957726577.png) 8. 在db_test库中执行如下语句，创建对应的测试表schema_test.table1。 ``` create table schema_test.table1(c1 int primary key,c2 varchar(10),c3 TIMESTAMP(6)); ``` #### 创建Kafka实例 本章节介绍创建Kafka实例。 1. 登录[华为云控制台](https://console.huaweicloud.com)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930500994.png)，选择区域。 3. 单击左侧的服务列表图标，选择"应用中间件 \> 分布式消息服务Kafka版"。 4. 单击"购买Kafka实例"。 5. 选择实例区域和可用区。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930660366.png) 6. 配置实例名称和实例规格等信息。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957619613.png) 7. 选择存储空间和容量阈值策略。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957699801.png) 8. 选择实例所属的VPC和安全组（[创建VPC](https://support.huaweicloud.com/bestpractice-drs/drs_04_0076.html#drs_04_0076__section1211181816438)和[创建安全组](https://support.huaweicloud.com/bestpractice-drs/drs_04_0076.html#drs_04_0076__section1969416454616)）。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930500998.png) 9. 配置实例密码。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930660370.png) 10. 单击"立即购买"，确认信息无误后单击"提交"。 11. 返回实例列表。 当Kafka实例运行状态为"运行中"时，表示实例创建完成。 #### 创建Kafka的Topic 1. 在"Kafka专享版"页面，单击Kafka实例的名称。 2. 选择"Topic管理"页签，单击"创建Topic"。 3. 在弹出的"创建Topic"的对话框中，填写Topic名称和配置信息，单击"确定"，完成创建Topic。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957619617.png) #### 创建DRS同步任务 1. 登录[华为云控制台](https://console.huaweicloud.com)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957726585.png)，选择区域。 3. 单击左侧的服务列表图标，选择"数据库 \> 数据复制服务 DRS"。 4. 选择左侧"实时同步管理"，单击"创建同步任务"。 5. 填写同步任务参数： 1. 配置同步任务名称。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930687142.png) 2. 选择需要同步任务的源库、目标数据库以及网络信息。 这里的源库选择[创建GaussDB主备实例]中创建的GaussDB实例。 图3同步实例信息 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000002106918346.png) 3. 选择规格类型和可用区。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930687146.png) 6. 单击"开始创建"。 同步实例创建中，大约需要5-10分钟。 7. 配置源库信息和目标库数据库密码。 1. 配置源库信息。 2. 单击"测试连接"。 当界面显示"测试成功"时表示连接成功。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957646401.png) 3. 选择目标库所在VPC和子网，填写Kafka的IP地址和端口。 4. 单击"测试连接"。 当界面显示"测试成功"时表示连接成功。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957726593.png) 8. 单击"下一步"。 9. 选择同步信息、策略、消息格式和对象等，投递到Kafka的消息格式。 本次选择如下。 ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930687150.png) 表2同步设置 | **类别** | **设置** | |:---|:---| | 同步Topic策略 | 集中投递到一个Topic，Topic名称"testTopic"。 不同Topica策略，对应选择的Kafka partition策略也不同，详细的说明可参考[同步Topic和Partition策略说明]。 | | 同步到Kafka partition策略 | 全部投递到同Partition 0。 不同Topica策略，对应选择的Kafka partition策略也不同，详细的说明可参考[同步Topic和Partition策略说明]。 | | 投递到Kafka的数据格式 | 可选择JSON格式，可参考[Kafka消息格式](https://support.huaweicloud.com/realtimesyn-drs/drs_03_0052.html)。 | | 同步对象 | 同步对象选择db_test库下的schema_test.table1。 | 10. 单击"下一步"，等待预检查结果。 11. 当所有检查都是"通过"时，单击"下一步"。 12. 确认同步任务信息正确后，单击"启动任务"。 返回DRS实时同步管理，查看同步任务状态。 启动中状态一般需要几分钟，请耐心等待。 当状态变更为"增量同步"，表示同步任务已启动。 ![](https://support.huaweicloud.com/bestpractice-drs/public_sys-resources/note_3.0-zh-cn.png) - 当前示例中GaussDB主备到Kafka选择单增量同步，任务启动后为增量同步状态。 - 增量同步会持续性同步增量数据，不会自动结束。 #### 确认同步任务执行结果 由于本次实践为增量同步模式，DRS任务会将源库的产生的增量数据持续同步至目标库中，直到手动任务结束。以下步骤通过在源库GaussDB中插入数据，查看Kafka的接收到的数据来验证同步结果。 1. 登录[华为云控制台](https://console.huaweicloud.com)。 2. 单击管理控制台左上角的![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001930527778.png)，选择区域。 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 ``` ![](https://support.huaweicloud.com/bestpractice-drs/zh-cn_image_0000001957646405.png) 9. 结束同步任务。 根据业务情况，确认数据已全部同步至目标库，可以结束当前任务。 1. 单击"操作"列的"结束"。 2. 仔细阅读提示后，单击"是"，结束任务。 #### 同步Topic和Partition策略说明 表3Topic和Partition策略说明 | Topic策略 | 对应可选的Partition策略 | 说明 | |:---|:---|:---| | **集中投递到一个Topic**： 对于源库业务量不大的场景，建议选择集中投递到一个Topic。 | 按库名.schema.表名的hash值投递到不同Partition | 适用于单表的查询场景，可以提高单表读写性能，推荐使用此选项。 | | **集中投递到一个Topic**： 对于源库业务量不大的场景，建议选择集中投递到一个Topic。 | 全部投递到Partition 0 | 适用于有事务要求的场景，写入性能比较差，如果没有强事务要求，不推荐使用此选项。 | | **集中投递到一个Topic**： 对于源库业务量不大的场景，建议选择集中投递到一个Topic。 | 按表的主键值hash值投递到不同的Partition | 适用于一个表一个Topic的场景，避免该表都写到同一个分区，消费者可以并行从各分区获取数据。 | | **集中投递到一个Topic**： 对于源库业务量不大的场景，建议选择集中投递到一个Topic。 | 按库名.schema的hash值投递到不同Partition | 适用于一个database一个topic的场景，避免多个schema下的数据都写到一个分区，消费者可以并行从各分区获取数据。 | | **集中投递到一个Topic**： 对于源库业务量不大的场景，建议选择集中投递到一个Topic。 | 按表的非主键列值的hash值投递到不同的Partition | 适用于一个表一个Topic的场景，避免该表都写到同一个分区，用户可以按照非主键列值的hash值自定义message key，消费者可以并行从各分区获取数据。 | | **按库名-schema-表名自动生成Topic名字**： 如果每张表数据量都非常大，建议选择自动生成Topic名字，按库名-schema-表名确定一个Topic。 | 全部投递到Partition 0 | 适用于有事务要求的场景，写入性能比较差，如果没有强事务要求，不推荐使用此选项。 | | **按库名-schema-表名自动生成Topic名字**： 如果每张表数据量都非常大，建议选择自动生成Topic名字，按库名-schema-表名确定一个Topic。 | 按表的主键值hash值投递到不同的Partition | 适用于一个表一个Topic的场景，避免该表都写到同一个分区，消费者可以并行从各分区获取数据。 | | **按库名-schema-表名自动生成Topic名字**： 如果每张表数据量都非常大，建议选择自动生成Topic名字，按库名-schema-表名确定一个Topic。 | 按表的非主键列值的hash值投递到不同的Partition | 适用于一个表一个Topic的场景，避免该表都写到同一个分区，用户可以按照非主键列值的hash值自定义message key，消费者可以并行从各分区获取数据。 | | **按库名自动生成Topic名字**： 对于源库数据量量不大的场景，可以选择一个database自动生成Topic名字，按库名确定一个Topic。 | 按库名.schema.表名的hash值投递到不同Partition | 适用于单表的查询场景，可以提高单表读写性能，推荐使用此选项。 | | **按库名自动生成Topic名字**： 对于源库数据量量不大的场景，可以选择一个database自动生成Topic名字，按库名确定一个Topic。 | 全部投递到Partition 0 | 适用于有事务要求的场景，写入性能比较差，如果没有强事务要求，不推荐使用此选项。 | | **按库名自动生成Topic名字**： 对于源库数据量量不大的场景，可以选择一个database自动生成Topic名字，按库名确定一个Topic。 | 按库名.schema的hash值投递到不同Partition | 适用于一个database一个topic的场景，避免多个schema下的数据写到一个分区，消费者可以并行从各分区获取数据。 | | **按库名-schema自动生成Topic名字**： 如果每个schema数据量都非常大，建议选择按库名-schema自动生成Topic名字，按库名-schema确定一个Topic。 | 按库名.schema.表名的hash值投递到不同Partition | 适用于单表的查询场景，可以提高单表读写性能，推荐使用此选项。 | | **按库名-schema自动生成Topic名字**： 如果每个schema数据量都非常大，建议选择按库名-schema自动生成Topic名字，按库名-schema确定一个Topic。 | 全部投递到Partition 0 | 适用于有事务要求的场景，写入性能比较差，如果没有强事务要求，不推荐使用此选项。 |