如何使用Maxwell进行全量表或增量表的数据同步
如何使用Maxwell进行全量表或增量表的数据同步
zzh1.Maxwell概述:
Maxwell是使用Java编写(可以使用jps查看进程)的MySQL变更数据抓取软件。他会实时监控Mysql数据库的数据变更操作（包括insert、update、delete），并将变更数据以JSON的格式发送给Kafka、Kinesi等流数据处理平台。
Maxwell输出数据格式
| 操作 | SQL语句 | 数据格式 |
|---|---|---|
| 插入 | insert into gmall.student values(1, 'zhangsan'); |
```json |
| { | ||
| “database”: “gmall”, | ||
| “table”: “student”, | ||
| “type”: “insert”, | ||
| “ts”: 1634004537, | ||
| “xid”: 1530907, | ||
| “commit”: true, | ||
| “data”: { | ||
| “id”: 1, | ||
| “name”: “zhangsan” | ||
| } | ||
| } | ||
| ``` | ||
| 更新 | update gmall.student set name = 'lisi' where id=1; |
```json |
| { | ||
| “database”: “gmall”, | ||
| “table”: “student”, | ||
| “type”: “update”, | ||
| “ts”: 1634004657, | ||
| “xid”: 1531916, | ||
| “commit”: true, | ||
| “data”: { | ||
| “id”: 1, | ||
| “name”: “lisi” | ||
| }, | ||
| “old”: { | ||
| “name”: “zhangsan” | ||
| } | ||
| } | ||
| ``` | ||
| 删除 | delete from gmall.student where id=1; |
```json |
| { | ||
| “database”: “gmall”, | ||
| “table”: “student”, | ||
| “type”: “delete”, | ||
| “ts”: 1634004751, | ||
| “xid”: 1532725, | ||
| “commit”: true, | ||
| “data”: { | ||
| “id”: 1, | ||
| “name”: “lisi” | ||
| } | ||
| } | ||
| ``` |
字段说明：
database：变更数据所属的数据库commit：事务提交标志，可用于重新组装事务。- xid：事务ID，用于标识事务。
- ts：变更数据的时间戳。
- type：变更数据的类型，包括insert、update、delete。
- data：变更数据，包括新增或修改的数据。
- old：变更前的数据，仅在update操作时存在。
2.Maxwell的实现原理
Maxwell的实现原理很简单，就是将自己伪装成Slave，并遵循Mysql主从复制的协议，从master中同步数据。
Mysql主从复制工作原理
① Master主库接收到数据变更请求，完成数据变更，并将其写到二级制日志（binary log）中。
② Slave从库向Mysql master发送dump协议，将Master主库的binary log events拷贝到从库的中继日志（relay log）中。
③ Slave从库读取并回放中继日志中的事件，将更新的数据同步到自己的数据库中。
3.Maxwell部署
1.配置MySQL
1.1
配置MySQL启动MySQL Binlog（修改MySQL的配置文件/etc/my.cnf重启Mysql服务后才生效）
2.1
Mysql Binlog模式介绍
① Statement-based: 基于语句，Binlog会记录所有写操作的SQL语句，包括：insert、update、delete等。
优点：节省空间；
缺点：有可能造成数据不一致，如：insert语句中包含now()函数。
② Row-based： 基于行，Binlog会记录每次写操作的被操作行记录的变化。
优点：保持数据的绝对一致性
缺点：占用较大空间
③ mixed： 混合模式，默认是Statement-based，如果SQL语句可能导致数据不一致，就自动切换到Row-based
Maxwell要求Binlog采用Row-based模式
2.创建Maxwell所需数据库和用户
3.配置Maxwell
3.1
修改Maxwell的配置文件config.properties,主要是配置数据发送的目的地。（若Maxwell发送数据的目的地是kafka集群，需要首先将kafka集群启动。）
4.启动Maxwell
Maxwell的启动命令：
1 | bin/maxwell --config config.properties --daemon |
Maxwell关闭命令：
1 | ps -ef | grep maxwell | grep -v grep | grep maxwell | awk '{print $2}' | xargs kill -9 |
Maxwell全量数据同步流程命令
1 | bin/maxwell-bootstrap --database edu --table user_info --config config.properties |