CentOS7环境Maxwell的安装及使用

Maxwell的安装

下载安装包

解压安装包

配置环境变量

启用MySQL Binlog

创建Maxwell所需数据库和用户

配置Maxwell

Maxwell的使用

启动Kafka集群

Maxwell启停

Maxwell启停脚本

MySQL数据准备

Kafka开启消费者

全量数据同步

增量数据同步

启动Kafka消费者

修改数据

添加数据

删除数据

查询数据

Maxwell是用Java编写的MySQL变更数据抓取软件。它会实时监控MySQL数据库的数据变更操作（insert、update、delete），并将变更数据以 JSON 格式发送给 Kafka、Kinesi等流数据处理平台。

Maxwell的安装

因为MySQL安装在node3机器（安装MySQL8），就近原则，所以Maxwell也安装在node3中。

下载安装包

官网下载安装包，下载版本如下：

maxwell-1.29.2.tar.gz

注意：Maxwell-1.30.0及以上版本不再支持JDK1.8。

解压安装包

[hadoop@node3 installfile]$ tar -zxvf maxwell-1.29.2.tar.gz -C ~/soft

配置环境变量

[hadoop@node3 installfile]$ sudo vim /etc/profile.d/my_env.sh

添加如下内容

#MAXWELL_HOME
export MAXWELL_HOME=/home/hadoop/soft/maxwell-1.29.2
export PATH=$PATH:$MAXWELL_HOME/bin

让环境变量生效

source /etc/profile

启用MySQL Binlog

MySQL的Binlog默认是未开启的，如需捕获更新操作，需要先进行开启Binlog。

修改MySQL配置文件/etc/my.cnf

[hadoop@node3 installfile]$ sudo vim /etc/my.cnf

增加如下配置

server-id = 1
log-bin=mysql-bin
binlog_format=row
binlog-do-db=gmall
binlog-do-db=test

配置项解释：

#数据库id
server-id = 1
#启动binlog，该参数的值会作为binlog的文件名
log-bin=mysql-bin
#binlog类型，maxwell要求为row类型
binlog_format=row
#启用binlog的数据库，需根据实际情况作出修改，如果需要开启多个数据库，直接再添加新的binlog-do-db设置行
binlog-do-db=gmall

MySQL Binlog类型：

Statement-based：基于语句，Binlog会记录所有写操作的SQL语句，包括insert、update、delete等。

优点：节省空间

缺点：有可能造成数据不一致，例如insert语句中包含now()函数。

Row-based：基于行，Binlog会记录每次写操作后被操作行记录的变化。

优点：保持数据的绝对一致性。

缺点：占用较大空间。

mixed：混合模式，默认是Statement-based，如果SQL语句可能导致数据不一致，就自动切换到Row-based。

Maxwell要求Binlog采用Row-based类型。

重启MySQL服务

[hadoop@node3 installfile]$ sudo systemctl restart mysqld

创建Maxwell所需数据库和用户

Maxwell需要在MySQL中存储其运行过程中的所需的一些数据，包括Binlog同步的断点位置（Maxwell支持断点续传）等等，故需要在MySQL为Maxwell创建数据库及用户。

1）创建数据库

[hadoop@node3 installfile]$ mysql -uroot -p000000
...
省略若干输出
...
msyql> CREATE DATABASE maxwell;

2）创建Maxwell用户并赋予其必要权限

mysql> CREATE USER 'maxwell'@'%' IDENTIFIED BY 'maxwell';
mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%';
mysql> GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

配置Maxwell

1）基于模板配置文件复制得到Maxwell配置文件

[hadoop@node3 installfile]$ cd ~/soft/maxwell-1.29.2

[hadoop@node3 maxwell-1.29.2]$ ls
bin        config.properties.example                    lib      log4j2.xml     README.md
config.md  kinesis-producer-library.properties.example  LICENSE  quickstart.md

[hadoop@node3 maxwell-1.29.2]$ cp config.properties.example config.properties

2）修改Maxwell配置文件

[hadoop@node3 maxwell-1.29.2]$ vim config.properties

配置如下

#Maxwell数据发送目的地，可选配置有stdout|file|kafka|kinesis|pubsub|sqs|rabbitmq|redis
producer=kafka
# 目标Kafka集群地址
kafka.bootstrap.servers=node2:9092,node3:9092,node4:9092
#目标Kafka topic，可静态配置，例如:maxwell，也可动态配置，例如：%{database}_%{table}
kafka_topic=topic_db# MySQL相关配置
host=node3
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true# 过滤gmall中的z_log表数据，该表是日志数据的备份，无须采集
filter=exclude:gmall.z_log
# 指定数据按照主键分组进入Kafka不同分区，避免数据倾斜
producer_partition_by=primary_key

Maxwell的使用

启动Kafka集群

若Maxwell发送数据的目的地为Kafka集群，则需要先确保Kafka集群为启动状态。

开启Kafka集群（Kafka集群安装）所有机器（node2、node3、node4），然后在任意一台机器执行如下命令：

[hadoop@node2 ~]$zk.sh start
[hadoop@node2 ~]$kf.sh start

Maxwell启停

1）启动Maxwell

maxwell --config $MAXWELL_HOME/config.properties --daemon

操作

[hadoop@node3 maxwell-1.29.2]$ maxwell --config $MAXWELL_HOME/config.properties --daemon
Redirecting STDOUT to /home/hadoop/soft/maxwell-1.29.2/bin/../logs/MaxwellDaemon.out
Using kafka version: 1.0.0
[hadoop@node3 maxwell-1.29.2]$ jps
1785 Maxwell
1839 Jps
[hadoop@node3 maxwell-1.29.2]$

2）停止Maxwell

ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9

操作

[hadoop@node3 maxwell-1.29.2]$ ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9
[hadoop@node3 maxwell-1.29.2]$ jps
1891 Jps
[hadoop@node3 maxwell-1.29.2]$

Maxwell启停脚本

创建并编辑Maxwell启停脚本

[hadoop@node3 maxwell-1.29.2]$ cd ~/bin
[hadoop@node3 bin]$ vim mxw.sh

脚本内容如下

#!/bin/bashMAXWELL_HOME=/home/hadoop/soft/maxwell-1.29.2status_maxwell(){result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`return $result
}start_maxwell(){status_maxwellif [[ $? -lt 1 ]]; thenecho "启动Maxwell"$MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemonelseecho "Maxwell正在运行"fi
}stop_maxwell(){status_maxwellif [[ $? -gt 0 ]]; thenecho "停止Maxwell"ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9elseecho "Maxwell未在运行"fi
}case $1 instart )start_maxwell;;stop )stop_maxwell;;restart )stop_maxwellstart_maxwell;;
esac

注意：MAXWELL_HOME需要根据实际情况修改。

赋予权限

[hadoop@node3 bin]$ chmod 777 mxw.sh

启动Maxwell

[hadoop@node3 bin]$ mxw.sh start

查看进程

[hadoop@node3 bin]$ jps
1988 Jps
1942 Maxwell

停止Maxwell

[hadoop@node3 bin]$ mxw.sh stop

查看进程

[hadoop@node3 bin]$ jps
2015 Jps

MySQL数据准备

进入mysql命令行，执行如下命令

create database test;
use test;
create table stu (id int, name varchar(100), age int);
insert into stu values(1,"张三",18);
insert into stu values(1,"李四",20);

Kafka开启消费者

在node2、node3、node4集群任意一台执行如下消费者命令，这里在node2执行

kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topic_db

全量数据同步

Maxwell提供了bootstrap功能来进行历史数据的全量同步，命令如下：

[hadoop@node3 bin]$ mxw.sh start

[hadoop@node3 bin]$ maxwell-bootstrap --database test --table stu --config $MAXWELL_HOME/config.properties

kafka消费者输出如下

{"database":"test","table":"stu","type":"bootstrap-start","ts":1719415852,"data":{}}
{"database":"test","table":"stu","type":"bootstrap-insert","ts":1719415852,"data":{"id":1,"name":"张三","age":18}}
{"database":"test","table":"stu","type":"bootstrap-insert","ts":1719415852,"data":{"id":1,"name":"李四","age":20}}
{"database":"test","table":"stu","type":"bootstrap-complete","ts":1719415852,"data":{}}

格式化后如下：

{"database": "test","table": "stu","type": "bootstrap-start","ts": 1719415852,"data": {}
} {"database": "test","table": "stu","type": "bootstrap-insert","ts": 1719415852,"data": {"id": 1,"name": "张三","age": 18}
} {"database": "test","table": "stu","type": "bootstrap-insert","ts": 1719415852,"data": {"id": 1,"name": "李四","age": 20}
} {"database": "test","table": "stu","type": "bootstrap-complete","ts": 1719415852,"data": {}
}

（1）第一条type为bootstrap-start和最后一条type为bootstrap-complete的数据，是bootstrap开始和结束的标志，不包含数据，中间的type为bootstrap-insert的数据才包含数据。

（2）一次bootstrap输出的所有记录的ts都相同，为bootstrap开始的时间(系统时间)。

增量数据同步

启动Kafka消费者

kafka-console-consumer.sh --bootstrap-server node2:9092 --topic topic_db

修改数据

将李四的id改为2

update stu set id=2 where name="李四";

kafka消费者输出

{"database":"test","table":"stu","type":"update","ts":1719416255,"xid":2595,"commit":true,"data":{"id":2,"name":"李四","age":20},"old":{"id":1}}

格式化输出

{"database": "test","table": "stu","type": "update","ts": 1719416255,"xid": 2595,"commit": true,"data": {"id": 2,"name": "李四","age": 20},"old": {"id": 1}
}

添加数据

例如：添加一条王五的数据

insert into stu values(3,"王五",23);

kafka消费者输出

{"database":"test","table":"stu","type":"insert","ts":1719416370,"xid":2853,"commit":true,"data":{"id":3,"name":"王五","age":23}}

格式化输出

{"database": "test","table": "stu","type": "insert","ts": 1719416370,"xid": 2853,"commit": true,"data": {"id": 3,"name": "王五","age": 23}
}

删除数据

delete from stu where id=3;

kafka消费者输出

{"database":"test","table":"stu","type":"delete","ts":1719416588,"xid":3339,"commit":true,"data":{"id":3,"name":"王五","age":23}}

格式化输出

{"database": "test","table": "stu","type": "delete","ts": 1719416588,"xid": 3339,"commit": true,"data": {"id": 3,"name": "王五","age": 23}
}

查询数据

select * from stu;

kafka消费者无新增的输出

可以看到，Maxwell可以监听到MySQL开启Binlog数据库的增、删、改操作。

完成！enjoy it！