支持的源和目标数据库
支持的源和目标数据库，如下表：
| 源数据库 | 目标数据库 |
|---|---|
| RDS for MySQL 自建 MySQL 5.6部分版本/5.7/8.0 | 自建Kafka 天翼云消息队列Kafka |
支持的迁移对象及SQL
迁移对象
当前DTS支持表级（指定的表对象）、库级（整库迁移）迁移。
数据迁移仅针对数据源中的用户数据库，而系统库会被自动过滤。例如：MySQL 数据源中的 information_schema、mysql、performance_schema、sys 库不会出现在可迁移列表当中。
若源库为空库（该库下未创建任何表），不支持作为待迁移对象。
增量迁移支持的SQL操作
DML
INSERT、UPDATE、DELETE。
DDL
常见DDL操作，如ALTER TABLE、TRUNCATE TABLE 和DROP TABLE。
投递到Kafka的数据格式
DTS投递到Kafka的消息支持JSON和Avro两种数据格式。
JSON格式
消息结构
JSON格式的消息结构如下：
| 参数 | 数据类型 | 说明 |
|---|---|---|
| dbType | String | 源库类型，如MYSQL、POSTGRESQL、SQLSERVER。 |
| id | Long | DTS内部定义的事件操作序列号，单调递增。 |
| es | Long | 源库产生数据的时间戳（毫秒）。全量数据时为0。 |
| ts | Long | 数据写入目标Kafka的时间戳（毫秒）。 |
| database | String | 源端数据库名称。 |
| table | String | 源端表名称。 |
| ddl | Boolean | 是否为DDL操作。 |
| sql | String | DDL操作的SQL语句，DML操作时取值为空字符串。 |
| opType | String | 操作类型。全量数据为INIT，增量DML为INSERT、UPDATE、DELETE，增量DDL为DDL。 |
| data | Array | 最新的数据，格式为键值对数组，每个元素为{"列名":"值"}。INSERT时为插入的数据；UPDATE时为更新后的数据；DELETE时为空数组。 |
| old | Array | 旧数据，格式同data。INSERT时为空数组；UPDATE时为更新前的数据；DELETE时为被删除的数据。 |
| pkNames | Array | 主键列名称列表。 |
| sqlType | Array | 源端表字段的JDBC类型编号，格式为键值对数组，每个元素为{"列名":类型编号}。 |
| oriType | Array | 源端表字段的原始类型名称，格式为键值对数组，每个元素为{"列名":"类型名"}。 |
| gtid | String | 增量日志位点。MySQL对应GTID，PostgreSQL对应LSN，SQL Server对应CDC的LSN。 |
JSON格式消息示例
全量数据（opType为INIT）
{"data":[{"id":"150"},{"username":"zhao6"},{"real_name":"赵六"},{"age":"33"},{"create_time":"2026-07-16 09:23:18.664747"},{"update_time":"2026-07-16 09:23:18.664747"}],"database":"test_schema","dbType":"POSTGRESQL","ddl":false,"es":0,"gtid":"","id":0,"old":[],"opType":"INIT","oriType":[{"id":"BIGINT"},{"username":"CHAR"},{"real_name":"CHAR"},{"age":"INTEGER"},{"create_time":"TIMESTAMP"},{"update_time":"TIMESTAMP"}],"pkNames":["id"],"sqlType":[{"id":-5},{"username":1},{"real_name":1},{"age":4},{"create_time":93},{"update_time":93}],"table":"t_user","ts":1784165589209}
增量INSERT
{"data":[{"id":"10"},{"username":"tom"},{"age":"23"},{"sex":"男"},{"phone":null},{"create_time":"2026-07-23 16:17:57.000"},{"home":null}],"database":"kxk10","dbType":"MySQL","ddl":false,"es":1784794677000,"gtid":"0@3639490@77940540@mysql-bin.000115@23863366@6ccd7c26-5974-11f1-8685-fa163e69b4f4:1-4944143@1784794677000","id":3639490,"old":[],"opType":"INSERT","oriType":[{"id":"INTEGER"},{"username":"CHAR"},{"age":"TINYINT"},{"sex":"CHAR"},{"phone":"CHAR"},{"create_time":"UNKNOWN"},{"home":"CHAR"}],"pkNames":["id"],"sqlType":[{"id":4},{"username":1},{"age":-6},{"sex":1},{"phone":1},{"create_time":4098},{"home":1}],"table":"user","ts":1784794678143}
增量UPDATE（普通字段变更，old中携带更新前的数据）
{"data":[{"id":"10"},{"username":"tom"},{"age":"23"},{"sex":"男"},{"phone":"13700137001"},{"create_time":"2026-07-23 16:17:57.000"},{"home":null}],"database":"kxk10","dbType":"MySQL","ddl":false,"es":1784795088000,"gtid":"0@3640726@77940540@mysql-bin.000115@24031046@6ccd7c26-5974-11f1-8685-fa163e69b4f4:1-4944555@1784795088000","id":3640726,"old":[{"id":"10"},{"username":"tom"},{"age":"23"},{"sex":"男"},{"phone":null},{"create_time":"2026-07-23 16:17:57.000"},{"home":null}],"opType":"UPDATE","oriType":[{"id":"INTEGER"},{"username":"CHAR"},{"age":"TINYINT"},{"sex":"CHAR"},{"phone":"CHAR"},{"create_time":"UNKNOWN"},{"home":"CHAR"}],"pkNames":["id"],"sqlType":[{"id":4},{"username":1},{"age":-6},{"sex":1},{"phone":1},{"create_time":4098},{"home":1}],"table":"user","ts":1784795089161}
增量UPDATE拆分投递（开启"UPDATE拆分投递"后，Partition Key变更时拆分为DELETE+INSERT两条消息）
第一条消息（DELETE，投递到原 Partition）：
{"data":[],"database":"kxk10","dbType":"MySQL","ddl":false,"es":1784795193000,"gtid":"0@3641044@77940540@mysql-bin.000115@24074182@6ccd7c26-5974-11f1-8685-fa163e69b4f4:1-4944661@1784795193000","id":3641044,"old":[{"id":"10"},{"username":"tom"},{"age":"23"},{"sex":"男"},{"phone":"13700137001"},{"create_time":"2026-07-23 16:17:57.000"},{"home":null}],"opType":"DELETE","oriType":[{"id":"INTEGER"},{"username":"CHAR"},{"age":"TINYINT"},{"sex":"CHAR"},{"phone":"CHAR"},{"create_time":"UNKNOWN"},{"home":"CHAR"}],"pkNames":["id"],"sqlType":[{"id":4},{"username":1},{"age":-6},{"sex":1},{"phone":1},{"create_time":4098},{"home":1}],"table":"user","ts":1784795193684}
第二条消息（INSERT，投递到新 Partition）：
{"data":[{"id":"110"},{"username":"tom"},{"age":"23"},{"sex":"男"},{"phone":"13700137001"},{"create_time":"2026-07-23 16:17:57.000"},{"home":null}],"database":"kxk10","dbType":"MySQL","ddl":false,"es":1784795193000,"gtid":"0@3641044@77940540@mysql-bin.000115@24074182@6ccd7c26-5974-11f1-8685-fa163e69b4f4:1-4944661@1784795193000","id":3641044,"old":[],"opType":"INSERT","oriType":[{"id":"INTEGER"},{"username":"CHAR"},{"age":"TINYINT"},{"sex":"CHAR"},{"phone":"CHAR"},{"create_time":"UNKNOWN"},{"home":"CHAR"}],"pkNames":["id"],"sqlType":[{"id":4},{"username":1},{"age":-6},{"sex":1},{"phone":1},{"create_time":4098},{"home":1}],"table":"user","ts":1784795193684}
增量DDL：
{"data":[{"ddl":"alter table user comment 'only for test'"},{"originDdl":"alter table user comment 'only for test'"}],"dbType":"MySQL","ddl":true,"es":1784795460000,"gtid":"0@3641850@77940540@mysql-bin.000115@24183431@6ccd7c26-5974-11f1-8685-fa163e69b4f4:1-4944930@1784795460000","id":3641850,"old":[],"opType":"DDL","oriType":[{"ddl":"NULL"},{"originDdl":"NULL"}],"pkNames":[],"sql":"alter table user comment 'only for test'","sqlType":[{"ddl":0},{"originDdl":0}],"ts":1784795461296}
Avro格式
Avro格式的消息使用DTS自定义的Avro Schema进行序列化，包含与JSON格式相同的字段信息，以二进制方式存储。
数据库账号及权限
数据库账号及权限如下表：
| 数据库 | 所需权限 | 参考赋权语句 |
|---|---|---|
| 源库 | 源库为MySQL5.7时：
源库为MySQL8.0时：
| 源库为MySQL5.7时： GRANT SELECT ON mysql.* TO '迁移账号'@'%'; GRANT SELECTON 待迁移的库.* TO '迁移账号'@'%'; GRANT RELOAD, REPLICATION CLIENT, REPLICATION SLAVE, PROCESS ON *.* TO '迁移账号'@'%';
源库为MySQL8.0时： GRANT SELECT ON mysql.* TO '迁移账号'@'%'; GRANT SELECT ON 待迁移的库.* TO '迁移账号'@'%'; GRANT RELOAD, PROCESS, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO '迁移账号'@'%'; |
| 目标库 | 具备目标Kafka集群的Producer写入权限。 |
操作须知
DTS迁移过程一般包含：预检查阶段、结构迁移阶段、全量阶段、增量阶段。为了确保同步各个阶段的平顺，在创建迁移任务前，请务必阅读以下使用须知。
如迁移对象为表级别，则单次迁移任务仅支持迁移最多10000张表（SQL Server为3000张）。当超出数量限制，任务会在提交后请求报错。如果遇到这种情形，建议您拆分待迁移的表，分批配置成多个任务，或者配置为整库迁移。
迁移过程中，请勿修改、删除提供给DTS连接访问的源库用户名、密码、权限，或修改源库的端口号。
选择表级对象迁移时，增量迁移过程中不建议对表进行重命名操作。
MySQL源数据库的binlog日志必须打开，且binlog日志格式必须为Row格式。
在磁盘空间允许的情况下，建议源数据库binlog保存时间越长越好，建议为7天。否则DTS在增量迁移时可能因无法获取Binlog而导致任务失败。
源数据库GTID状态建议为开启状态，源数据库实例没有开启GTID的情况下DTS不支持主备HA切换。
增量迁移时，必须设置MySQL源数据库的server_id。如果源数据库版本小于或等于MySQL5.6，server_id的取值范围在2－4294967296之间；如果源数据库版本大于或等于MySQL5.7，server_id的取值范围在1－4294967296之间。
操作步骤
1、购买DTS数据迁移实例。
在管理控制台点击"创建实例"进入订购页面，"实例类型"选择"数据迁移"，"目标库实例"的"数据库类型"选择Kafka，选择实例或输入IP地址、端口，完成其他信息的填写并完成购买。
2、进入实例配置页面。
DTS实例购买成功后，进入【数据迁移】实例列表页面，上一步骤购买成功的实例在实例列表中显示状态为“待配置”，进入实例配置页面的操作分两种情况：
当源库或目标库的网络接入类型为“公网EIP”时，请在实例列表的“IP地址列”点击“绑定弹性IP”按钮为DTS实例完成公网弹性IP的绑定，然后点击该实例操作列的“实例配置”按钮。
当源库和目标库的网络接入类型不是“公网EIP”时，请直接点击该实例操作列的“实例配置”按钮。
3、配置源库及目标库信息。
进入实例配置第一个步骤的【配置源库及目标库信息】页面，填入源库与目标库的相关配置信息：
源库：选择数据库类型MySQL，输入源库IP地址、端口、数据库账号、数据库密码等信息。
目标库：选择数据库类型为Kafka，输入Kafka集群的Bootstrap Servers地址。若Kafka开启了SASL认证，需选择安全协议（SASL_PLAINTEXT）并填写认证机制及账密信息。
完成上述信息的填写后，点击源数据库和目标数据库的"测试连接"按钮进行数据库连接测试，检查数据库能否正常连接。
说明
目标库为Kafka的链路，支持的认证方式：
安全协议为PLAINTEXT时，无需账密
安全协议为SASL_PLAINTEXT时，支持如下SASL机制：
SASL机制为GSSAPI
SASL机制为PLAIN，需账密
SASL机制为SCRAM-SHA-256，需账密(加密，更安全)
SASL机制为SCRAM-SHA-512，需账密(加密，更安全)
注意：当前天翼云Kafka实例暂不支持SASL机制为SCRAM-SHA-256
4、配置迁移对象及高级配置。
源库和目标库连通性测试成功后，点下一步按钮，进入实例配置第二个步骤的【配置迁移对象及高级配置】页面，选择Topic映射策略、Partition投递策略、是否拆分Update、压缩算法、DDL投递Topic等参数，并在"源库对象"中选择要迁移的源库对象，包含：库、TABLE等，选中后点击">"按钮，将待迁移对象移动到"已选择对象"中。
在"已选择对象"窗口，可以点击表名后的“编辑”按钮，进入“编辑表”弹窗，支持设置WHERE条件过滤数据，只有满足WHERE条件的数据才会被同步到目标Topic。过滤条件支持标准的SQL WHERE语句（仅支持=、!=、<和>操作符）。
重要配置项说明：
| 配置项 | 可选项 | 说明 | 备注 |
|---|---|---|---|
Topic映射策略 | 自动生成Topic名字 | 按照固定格式自动生成Topic名称。支持使用database和tablename占位符自定义生成规则，例如：database-tablename-example将生成形如test-user-example的Topic名称。 | |
集中投递到固定的Topic | 用户选择目标Kafka中已存在的Topic。整库迁移时，所有表的数据将投递到同一个Topic；非整库迁移时，可为每张表指定不同的Topic。 | ||
| 同步到Kafka partition策略 | 按主键的Hash值投递到不同Partition | DTS将根据主键（或自定义Partition Key）计算Hash值，将不同的行投递到目标Topic的各Partition中。支持自定义Partition Key，指定单个或多个列作为Partition Key来计算Hash值。 注意： 当选择"按主键的Hash值投递到不同Partition"策略时，支持自定义Partition Key。您可以指定单个或多个列作为Partition Key来计算Hash值，DTS将根据计算得到的Hash值将不同的行投递到目标Topic的各Partition中。若未自定义Partition Key，则默认使用主键/唯一键；无任何键时使用全字段做Hash。 | |
按表名Hash投递到不同Partition | DTS将根据表名计算Hash值进行Partition路由，保证同一张表的数据有序。 | ||
全部投递到Partition 0 | 所有数据均投递到目标Topic的Partition 0中。 | ||
| 更新分区键后拆分消息投递 | 是 | 当Partition Key的值被修改时，DTS会往原来路由的Partition投递一条opType为DELETE的消息，再根据Partition Key的新值往新的Partition投递一条opType为INSERT的消息，确保下游消费者能正确感知数据路由变化。 | 仅当同步到Kafka partition策略配置项选择”按主键的Hash值投递到不同Partition“时，才需要配置该选项。 |
| 否（默认值） | Partition Key变更时，只会根据Partition key的新值往新的partition投递一条opType为Update的消息。 | ||
| Kafka压缩格式 | gzip | 使用gzip压缩算法。 | DTS支持对投递到Kafka中的消息按选择的压缩算法进行压缩。 |
snappy | 使用snappy压缩算法。 | ||
lz4 | 使用lz4压缩算法。 | ||
| 存储DDL的Topic | 指定Topic | 按配置的目标Topic存储增量迁移过程中产生的DDL信息。 | |
| 不指定Topic（默认值） | 增量迁移过程中产生的DDL信息，默认投递到dts-inc-ddl这个Topic中。 |
5、配置库表列信息
当在【配置迁移对象及高级配置】页面的同步到Kafka partition策略配置项选择”按主键的Hash值投递到不同Partition“时，完成迁移对象和高级配置后，点击“下一步”，进入【配置库表列】页面。在这里可以选择往Kafka投递数据时使用的分区键，默认为源端表的主键。
6、预检查和启动迁移
完成迁移对象和高级配置后，点击"下一步预检查"，进入实例配置第三个步骤的【预检查】页面。预检查会检查如下列表信息，并给出检查结果，用户可以依据检查结果进行下一步操作。
| 检查项 | 检查内容 |
|---|---|
| 源库连通性检查 | 检查数据传输服务能否连通源数据库。 |
| 目标库连通性检查 | 检查数据传输服务能否连通目标Kafka集群。 |
| 源库用户权限检查 | 检查源库用于DTS任务的用户是否具有相应的权限。 |
| 待迁移表主键检查 | 检查待迁移表是否都存在主键。 |
| 源库binlog是否开启检查 | 查看源库的log_bin参数是不是ON。 |
| 源库binlog模式检查 | 查看源库的binlog_format参数是不是ROW。 |
| 源库binlog影像类型检查 | 查看源库的binlog_row_image参数是不是FULL。 |
| 源库binlog存在性检查 | 查看源库的binlog文件是否被误删除。 |
| 源库binlog保留时间检查 | 检查源库的binlog保留时间是否满足要求。 |
| 存储引擎检查 | 检查源库中待迁移的表的存储引擎。 |
如果预检查通过，可点击【预检查】页面底部的“启动迁移”按钮，开始迁移任务。
