大数据Maxwell（二）：使用Maxwell增量和全量同步MySQL数据
使用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=kafka
kafka.bootstrap.servers=node1:9092,node2:9092,node3:9092
kafka_topic=maxwell_topic
#设置根据表将binlog写入Kafka不同分区，还可指定：[database, table, primary_key, transaction_id, thread_id, column]
producer_partition_by=table
#mysql 节点
host=node2
#连接mysql用户名和密码
user=maxwell
password=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_topic5、启动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_first2、在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中。
- 点赞
- 收藏
- 关注作者
评论（0）