苏州建设工程招标网乐陵310seo
目录
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_keyMaxwell的使用
启动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！