# MySQL 迁移到 Kafka
UDTS 支持 从 MySQL 迁移到 Kafka。
 MySQL 支持版本有 MySQL(包含Percona版)5.6/5.7/8.x。
 Kafka 支持版本 2.x：包括 2.0、2.1、2.2、2.3、2.4、2.5、2.6、2.7、2.8 等。

## 1. 功能限制

### 1.1 源MySQL限制
1. 增量/全+增迁移时，源库需要开启binlog，且格式设置为ROW, image设置为FULL。

```

查询方式:
show global variables like 'binlog_format';
show global variables like 'binlog_row_image';

设置方式：
set global binlog_format = "ROW" ;
set global binlog_row_image = "FULL" ;

```

### 1.2 目标Kafka限制

1. 设置 `auto.create.topics.enable` 为true。 
2. 设置`delete.topic.enable` 为true


## 2. 迁移内容

| 迁移内容 | 说明                                                                |
| -------- | ------------------------------------------------------------------- |
| 迁移结构 | Database、Table 结构及数据                                          |
| 迁移范围 | 仅迁移创建任务时可以查到的库表， 任务运行中新增的表暂时不会自动迁移 |
| DDL      | CREATE、ALTER、DROP 等语句                                          |
| DML      | Snapshot/insert/update/delete                                       |


## 3. 表单填写

### 数据源表单

| 参数名   | 说明                                                                                                                                                                                      |
| -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| 地址类型 | 支持内网地址，外网地址，专线地址三种方式。内网地址需要填写VPC和子网信息；外网地址支持IP和域名两种方式；专线地址既支持IP，也支持域名，如果使用域名需要用户网络有外网出口。                 |
| 端口     | MySQL连接端口                                                                                                                                                                             |
| 用户名   | MySQL连接用户名                                                                                                                                                                           |
| 密码     | MySQL数据库对应用户密码                                                                                                                                                                   |
| 数据库名 | MySQL数据库名称。 所有库传输请填写 *； 指定一个数据库传输，请填写数据库名；指定多个数据库传输，依次输入多个数据库名，库名之间使用英文逗号隔开。(如果数据库名称中包含空格则无法做增量迁移) |  |
| 表名     | MySQL传输表名。 只有当“数据库名”为指定一个数据库时有效。 若不填，默认为迁移指定库中的所有表； 指定一张表传输， 请填写表名； 指定多张表传输，依次输入多张表名，表名之间使用英文逗号隔开    |
| Nolocks  | 默认关闭，对于无法获取 SUPER 权限的友商RDS服务需要开启，UDB获取 SUPER 权限详见FAQ                                                                                                         |


###  传输目标表单

| 参数名       | 说明                                                                                                          |
| ------------ | ------------------------------------------------------------------------------------------------------------- |
| 内网地址     | Kafka集群的连接地址，示例：92.168.1.10:9092,192.168.1.11:9092,192.168.1.12:9093                               |
| 最大速率     | Kafka传输限速值，调整传输速率                                                                                 |
| Topic前缀    | UDTS迁移MySQL到Kafka会在目标Kafka上创建对应的Topic， 规则是每张表创建一个Topic，每个Topic都会以此参数作为前缀 |
| 默认分区数量 | UDTS迁移MySQL到Kafka会在目标Kafka上创建对应Topic的默认分区数量                                                |


## 4. Kafka传输数据格式

UDTS 迁移到Kafka的内容，格式为 Debezium JSON 格式, 下游同步可使用Flink CDC ，数据格式指定为 debezium-json。