【云小课】【第40课】如何通过DRS实现RDS for MySQL到Kafka的数据同步

数据复制服务（DRS）是一种易用、稳定、高效、用于数据同步的云服务，本节小课为您介绍，如何通过DRS将RDS for MySQL实例的增量数据同步到分布式消息服务Kafka。

使用场景

DRS实时同步功能一般用于建立数据同步通道，解决数据共享问题，也可以用于数据流式集成，具有数据转换能力，如库表映射，行列过滤等。本实践中的选择均为测试简化基本操作，仅做参考，实际情况请用户按业务场景选择，更多关于DRS的使用场景请单击这里了解。

部署架构

本示例中，DRS源数据库为华为云RDS for MySQL，目标端为华为云同Region下的分布式消息服务Kafka，通过VPC网络，将源数据库的增量数据同步到目标端。

更多关于DRS的使用场景请单击这里了解。

源端RDS for MySQL准备

创建RDS for MySQL实例

如何创建RDS for MySQL实例，请点击这里查看详细步骤。

构造数据

1. 登录华为云控制台。

2. 单击管理控制台左上角的，选择区域“华南-广州”。

3. 单击左侧的服务列表图标，选择“数据库 > 云数据库 RDS”。

4. 选择RDS实例，单击实例后的“更多 > 登录”。

5. 在弹出的对话框中单击“测试连接”检查。

6. 连接成功后单击“登录”。

7. 输入实例密码，登录RDS实例。

8. 单击“新建数据库”，创建db_test测试库。

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

CREATE TABLE `db_test`.`table3_` ( `Column1` INT(11) UNSIGNED NOT NULL, `Column2` TIME NULL, `Column3` CHAR NULL, PRIMARY KEY (`Column1`) ) ENGINE = InnoDB DEFAULT CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci;

目标端Kafka准备

创建Kafka实例

1. 登录华为云控制台。

2. 单击管理控制台左上角的，选择区域“华南-广州”。

3. 单击左侧的服务列表图标，选择“应用中间件 > 分布式消息服务Kafka版”。

4. 单击“购买Kafka实例”。

5. 选择实例区域和可用区。

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

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

8. 选择实例所属的VPC和安全组。

VPC和安全组已在创建VPC和安全组中准备好。

9. 配置实例密码。

10. 单击“立即购买”。

11. 返回实例列表。

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

创建Topic

1. 在“Kafka专享版”页面，单击Kafka实例的名称。

2. 选择“Topic管理”页签，单击“创建Topic”。

3. 在弹出的“创建Topic”的对话框中，填写Topic名称和配置信息，单击“确定”，完成创建Topic。

创建DRS同步任务

本章节介绍创建DRS实例，将RDS for MySQL上的数据库增量同步到Kafka。

同步前检查

在创建任务前，需要针对同步条件进行手工自检，以确保您的同步任务更加顺畅。

本示例中，为RDS for MySQL到Kafka的出云同步，您可以参考出云同步使用须知获取相关信息。

操作步骤

介绍RDS for MySQL到Kafka增量同步任务的详细操作过程。

1. 登录华为云控制台。

2. 单击管理控制台左上角的，选择区域“华南-广州”。

3. 单击左侧的服务列表图标，选择“数据库 > 数据复制服务 DRS”。

4. 选择左侧“实时同步管理”，单击“创建同步任务”。

5. 填写同步任务参数：

配置同步任务名称。

选择需要同步任务的源库、目标数据库以及网络信息。这里的目标库选择源端RDS for MySQL准备创建的RDS实例。

企业项目选择“default”。

6. 单击“下一步”。

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

7. 配置源库信息和目标库数据库密码。

配置源库信息。

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

选择目标库所在VPC和子网，填写Kafka的IP地址和端口。

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

8. 单击“下一步”。

9. 选择同步信息、策略、消息格式和对象等，投递到Kafka的消息格式。

本次选择如下。

类别

设置

同步Topic策略

集中投递到一个Topic，Topic名称“testTopic”。

同步到Kafka partition策略

按表名+库名的hash值投递到不同Partition。

投递到Kafka的数据格式

可选择JSON格式，可参考Kafka消息格式。

同步对象

同步对象选择db_test下的table3_表。

10.单击“下一步”。

11. 选择数据加工方式。RDS for MySQL到Kafka数据同步目前只支持列加工，列加工提供列级的查询和过滤能力。

12. 单击“下一步”，等待预检查结果。

13. 当所有检查都是“通过”时，单击"下一步”。

14. 确认同步任务信息正确后，单击“启动任务”。

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

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

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

说明：

目前RDS for MySQL到Kafka仅支持增量同步，任务启动后为增量同步状态。

如果创建的任务为全量同步，任务启动后进行全量数据同步，数据同步完成后任务自动结束。

如果创建的任务为全量+增量同步，任务启动后先进入全量同步，全量数据同步完成后进入增量同步状态。

增量同步会持续性同步增量数据，不会自动结束。

确认同步任务执行结果

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

操作步骤

1. 登录华为云控制台。

2. 单击管理控制台左上角的，选择区域“华南-广州”。

3. 单击左侧的服务列表图标，选择“数据库 > 云数据库 RDS””。

4. 单击RDS实例后的“更多 > 登录”。

5. 在弹出的对话框中单击“测试连接”检查。

6. 连接成功后单击“登录”。

7. 输入实例密码，登录RDS实例。

8. 在DRS同步对象的db_test.table3_表中，执行如下语句，插入数据。

INSERT INTO `db_test`.`table3_` (`Column1`,`Column2`,`Column3`) VALUES(4,'00:00:44','ddd');

9. 单击左侧的服务列表图标，选择“应用中间件 > 分布式消息服务Kafka版”。

10. 在“Kafka专享版”页面，单击Kafka实例的名称。

11. 选择“消息查询”页签，在Kafka对应的Topic中，查看接收到相应的JSON格式数据。

12. 结束同步任务。

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

单击“操作”列的“结束”。

仔细阅读提示后，单击“是”，结束任务。

云小课 数据复制服务 DRS 数据库

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。