文章转自华为云社区，作者：数据库的小云妹；原文链接：https://bbs.huaweicloud.com/blogs/329925
数据复制服务（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实例，请点击这里查看详细步骤。
构造数据
-
登录华为云控制台。
-
单击管理控制台左上角的
，选择区域“华南-广州”。 -
单击左侧的服务列表图标，选择“数据库 > 云数据库 RDS”。
-
选择RDS实例，单击实例后的“更多 > 登录”。
-
在弹出的对话框中单击“测试连接”检查。
-
连接成功后单击“登录”。
-
输入实例密码，登录RDS实例。
-
单击“新建数据库”，创建db_test测试库。
-
在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实例
-
登录华为云控制台。
-
单击管理控制台左上角的，选择区域“华南-广州”。
-
单击左侧的服务列表图标，选择“应用中间件 > 分布式消息服务Kafka版”。
-
单击“购买Kafka实例”。
-
选择实例区域和可用区。
-
配置实例名称和实例规格等信息。
-
选择存储空间和容量阈值策略。
-
选择实例所属的VPC和安全组。
VPC和安全组已在创建VPC和安全组中准备好。 -
配置实例密码。
-
单击“立即购买”。
-
返回实例列表。
当Kafka实例运行状态为“运行中”时，表示实例创建完成。
创建Topic
-
在“Kafka专享版”页面，单击Kafka实例的名称。
-
选择“Topic管理”页签，单击“创建Topic”。
-
在弹出的“创建Topic”的对话框中，填写Topic名称和配置信息，单击“确定”，完成创建Topic。
创建DRS同步任务
本章节介绍创建DRS实例，将RDS for MySQL上的数据库增量同步到Kafka。
同步前检查
在创建任务前，需要针对同步条件进行手工自检，以确保您的同步任务更加顺畅。
本示例中，为RDS for MySQL到Kafka的出云同步，您可以参考出云同步使用须知获取相关信息。
操作步骤
介绍RDS for MySQL到Kafka增量同步任务的详细操作过程。
-
登录华为云控制台。
-
单击管理控制台左上角的，选择区域“华南-广州”。
-
单击左侧的服务列表图标，选择“数据库 > 数据复制服务 DRS”。
-
选择左侧“实时同步管理”，单击“创建同步任务”。
-
填写同步任务参数：
- 配置同步任务名称。
- 选择需要同步任务的源库、目标数据库以及网络信息。这里的目标库选择源端RDS for MySQL准备创建的RDS实例。
- 企业项目选择“default”。
- 单击“下一步”。
同步实例创建中，大约需要5-10分钟。
- 配置源库信息和目标库数据库密码。
- 配置源库信息。
- 单击“测试连接”。当界面显示“测试成功”时表示连接成功。
- 选择目标库所在VPC和子网，填写Kafka的IP地址和端口。
- 单击“测试连接”。当界面显示“测试成功”时表示连接成功。
-
单击“下一步”。
-
选择同步信息、策略、消息格式和对象等，投递到Kafka的消息格式。
本次选择如下。
表1 同步设置
| 类别 | 设置 |
|---|---|
| 同步Topic策略 | 集中投递到一个Topic，Topic名称“testTopic”。 |
| 同步到Kafka partition策略 | 按表名+库名的hash值投递到不同Partition。 |
| 投递到Kafka的数据格式 | 可选择JSON格式，可参考Kafka消息格式。 |
| 同步对象 | 同步对象选择db_test下的table3_表。 |
-
单击“下一步”。
-
选择数据加工方式。RDS for MySQL到Kafka数据同步目前只支持列加工，列加工提供列级的查询和过滤能力。
-
单击“下一步”，等待预检查结果。
-
当所有检查都是“通过”时，单击"下一步”。
-
确认同步任务信息正确后，单击“启动任务”。
返回DRS实时同步管理，查看同步任务状态。
启动中状态一般需要几分钟，请耐心等待。
当状态变更为“增量同步”，表示同步任务已启动。
说明：
- 目前RDS for MySQL到Kafka仅支持增量同步，任务启动后为增量同步状态。
- 如果创建的任务为全量同步，任务启动后进行全量数据同步，数据同步完成后任务自动结束。
- 如果创建的任务为全量+增量同步，任务启动后先进入全量同步，全量数据同步完成后进入增量同步状态。
- 增量同步会持续性同步增量数据，不会自动结束。
确认同步任务执行结果
由于本次实践为增量同步模式，DRS任务会将源库的产生的增量数据持续同步至目标库中，直到手动任务结束。下面我们通过在源库RDS for MySQL中插入数据，查看Kafka的接收到的数据来验证同步结果。
操作步骤
-
登录华为云控制台。
-
单击管理控制台左上角的，选择区域“华南-广州”。
-
单击左侧的服务列表图标，选择“数据库 > 云数据库 RDS””。
-
单击RDS实例后的“更多 > 登录”。
-
在弹出的对话框中单击“测试连接”检查。
-
连接成功后单击“登录”。
-
输入实例密码，登录RDS实例。
-
在DRS同步对象的db_test.table3_表中，执行如下语句，插入数据。
INSERT INTO `db_test`.`table3_` (`Column1`,`Column2`,`Column3`) VALUES(4,'00:00:44','ddd');
-
单击左侧的服务列表图标，选择“应用中间件 > 分布式消息服务Kafka版”。
-
在“Kafka专享版”页面，单击Kafka实例的名称。
-
选择“消息查询”页签，在Kafka对应的Topic中，查看接收到相应的JSON格式数据。
-
结束同步任务。
根据业务情况，确认数据已全部同步至目标库，可以结束当前任务。
- 单击“操作”列的“结束”。
- 仔细阅读提示后，单击“是”，结束任务。
