将MySQL迁移到Kafka 本页为使用数据传输服务DTS将MySQL数据迁移到Kafka的详细介绍,包括数据迁移支持的源库、目标库,支持的迁移对象及SQL,数据库账号权限说明,操作须知,操作步骤等。 支持的源和目标数据库 支持的源和目标数据库,如下表: 源数据库 目标数据库 RDS for MySQL 自建 MySQL 5.6部分版本/5.7/8.0 自建Kafka 天翼云消息队列Kafka 支持的迁移对象及SQL 迁移对象 当前DTS支持表级(指定的表对象)、库级(整库迁移)迁移。 数据迁移仅针对数据源中的用户数据库,而系统库会被自动过滤。例如:MySQL 数据源中的 informationschema、mysql、performanceschema、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。