​使用Maxwell增量和全量同步MySQL数据

一、使用Maxwell同步MySQL数据

首先下载Maxwell,Maxwell下载地址：

(相关资料图)

https://github.com/zendesk/maxwell/releases/tag/v1.28.2

现在版本1.30.0版本之上需要jdk11以上，建议下载1.30.0版本以下版本。下载完成后按照如下步骤进行配置，同步MySQL数据到Kafka，前提是MySQL需要开启Binlog日志，可以参考Canal章节设置。Maxwell不支持高可用搭建，但是支持断点还原，可以在执行失败时重新启动继续上次位置读取数据。

1、将下载好的安装包上传到node3并解压

[root@node3 ~]# cd /software/[root@node3 software]# tar -zxvf ./maxwell-1.28.2.tar.gz 

2、在MySQL中创建Maxwell的用户及赋权

Maxwell同步mysql数据到Kafka中需要将读取的binlog位置文件及位置信息等数据存入MySQL，所以这里创建maxwell数据库，及给maxwell用户赋权访问其他所有数据库。

3、修改配置“config.properties”文件

进入“/software/maxwell-1.28.0”，修改“config.properties.example”为“config.properties”并配置：

producer=kafkakafka.bootstrap.servers=node1:9092,node2:9092,node3:9092kafka_topic=maxwell_topic#设置根据表将binlog写入Kafka不同分区，还可指定：[database, table, primary_key, transaction_id, thread_id, column]producer_partition_by=table#mysql 节点host=node2#连接mysql用户名和密码user=maxwellpassword=maxwell

注意：以上参数也可以在后期启动maxwell时指定参数方式来设置。

4、启动zookeeper及Kafka,并监控Kafka maxwell_topic

[root@node2 bin]# cd /software/kafka_2.11-0.11/[root@node2 bin]# ./kafka-console-consumer.sh  --bootstrap-server node1:9092,node2:9092,node3:9092 --topic maxwell_topic

5、启动Maxwell

[root@node3 ~]# cd /software/maxwell-1.28.2/bin[root@node3 bin]# maxwell --config ../config.properties

注意以上启动也可以编写脚本：

#startMaxwell.sh 脚本内容：/software/maxwell-1.28.2/bin/maxwell --config /software/maxwell-1.28.2/config.properties > ./log.txt 2>&1 &

修改执行权限：

chmod +x ./start_maxwell.sh 

6、向MySQL中增删改查写入数据

mysql> create database mysqldb;mysql> use mysqldb;mysql> create table info(id int,name varchar(255),age int);mysql> insert into info values (10,"xx",20);mysql> update info set age = 100 where id = 10;mysql> delete from info where id = 10;#对应Kafka中的消息如下：{"database":"mysqldb","table":"info","type":"insert","ts":1619000098,"xid":3890,"commit":true,"data":{"id":100,"name":"aaa","age":10}}{"database":"mysqldb","table":"info","type":"update","ts":1619000152,"xid":4142,"commit":true,"data":{"id":100,"name":"aaa","age":20},"old":{"age":10}}{"database":"mysqldb","table":"info","type":"delete","ts":1619000183,"xid":4228,"commit":true,"data":{"id":100,"name":"aaa","age":20}}

7、测试Maxwell断点续传

停止Maxwell,向MySQL中插入新的数据，重启Maxwell观察是否从上次消费到的binlog位置继续消费。

#使用kill -9 xxx 命令在node3停止Maxwell#向MySQL继续插入数据mysql> insert into info values (200,"bbb",20);mysql> update info set age = 30 where id = 200;#重新在node3启动Maxwell，可以观察到Kafka中继续上次binlog位置写入数据{"database":"mysqldb","table":"info","type":"insert","ts":1619000378,"xid":4565,"commit":true,"data":{"id":200,"name":"bbb","age":20}}{"database":"mysqldb","table":"info","type":"update","ts":1619000391,"xid":4566,"commit":true,"data":{"id":200,"name":"bbb","age":30},"old":{"age":20}}

二、使用Maxwell Bootstrap全量同步MySQL数据

Maxwell Bootstrap可以将MySQL中已经存在的数据批量同步到Kafka中，操作步骤如下：

1、修改“/software/maxwell-1.28.2/config.properties”配置文件

停止maxwell进程，在当前config.properties配置文件最后一行添加配置“client_id”,此配置项是指定当前maxwell启动后连接mysql的实例id,名字自取，在全量同步数据时需要使用到。

#指定maxwell 当前连接mysql的实例id,名字自取client_id=maxwell_first

2、在mysql中创建库“mysqldb2”,并插入数据

mysql> create database mysqldb2;mysql> use mysqldb2;mysql> create table t1 (id int ,name varchar(255),age int );mysql> insert into t1 values (1,"zs",18),(2,"ls",19),(3,"ww",20);

3、重新启动Maxwell,然后启动maxwell-bootstrap全量导数据

maxwell-bootstrap脚本可以指定MySQL数据库及表参数，同步MySQL指定库下对应表的全量数据，同时可以指定where条件。

[root@node3 ~]# cd /software/maxwell-1.28.2/bin#重启maxwell[root@node3 bin]# maxwell --config ../config.properties#同步mysqldb2 .t1表的全量数据[root@node3 bin]# ./maxwell-bootstrap --database mysqldb2 --table t1 --host node2 --user maxwell --password maxwell  --client_id maxwell_first --where "id<=2"

注意：maxwell-bootstrap命令执行后，可以在对应的topic中查看到数据，这里对应的topic是maxwell进程对应config.properties文件中配置的topic。同时maxwell-bootstrap命令指定的client_id 需要与maxwell进行对应的config.properties配置文件中配置的一样。--where是指定条件，只会全量导入满足条件的数据，有了where条件可以使maxwell-bootstrap进程配合maxwell实时同步进程将一张表数据无缝同步到Kafka中。

关键词：