Maxwell 介绍

Maxwell 是由美国Zendesk开源，用Java编写的MySQL实时抓取软件。 实时读取MySQL二进制日志Binlog，并生成 JSON 格式的消息，作为生产者发送给 Kafka，Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件或其它平台的应用程序。

官网地址：http://maxwells-daemon.io/

Maxwell 工作原理

Mysql主从复制

• Master主库将改变记录，写到二进制日志(binary log)中
• Slave从库向mysql master发送dump协议，将master主库的binary log events拷贝到它的中继日志(relay log)；
• Slave从库读取并重做中继日志中的事件，将改变的数据同步到自己的数据库。

Maxwell的工作原理

很简单，就是把自己伪装成slave，假装从master复制数据

MySQL的Binlog

• 找到MySQL配置文件的位置

Linux: /etc/my.cnf
Windows: \my.ini

如果/etc目录下没有，可以通过locate my.cnf查找位置

• 在mysql的配置文件下,修改配置

在[mysqld] 区块，设置/添加  log-bin=mysql-bin

（1）这个表示binlog日志的前缀是mysql-bin，以后生成的日志文件就是 mysql-bin.123456 的文件后面的数字按顺序生成

（2）每次mysql重启或者到达单个文件大小的阈值时，新生一个文件，按顺序编号。

• binlog的分类设置

mysql binlog的格式有三种，分别是STATEMENT,MIXED,ROW。

在配置文件中可以选择配置binlog_format= statement|mixed|row

三种格式的区别：

（1）statement

语句级：binlog会记录每次一执行写操作的语句。

相对row模式节省空间，但是可能产生不一致性，比如

update tt set create_date=now()

如果用binlog日志进行恢复，由于执行时间不同可能产生的数据就不同。

优点：节省空间

缺点：有可能造成数据不一致。

（2）row

行级： binlog会记录每次操作后每行记录的变化。

优点：保持数据的绝对一致性。因为不管sql是什么，引用了什么函数，他只记录执行后的效果。

缺点：占用较大空间。

（3）mixed

statement的升级版，一定程度上解决了，因为一些情况而造成的statement模式不一致问题

默认还是statement，在某些情况下譬如：

（1）当函数中包含 UUID() 时；
（2）包含 AUTO_INCREMENT 字段的表被更新时；
（3）执行 INSERT DELAYED 语句时；
（4）用 UDF 时；

会按照 ROW的方式进行处理

优点：节省空间，同时兼顾了一定的一致性。

缺点：还有些极个别情况依旧会造成不一致，另外statement和mixed对于需要对binlog的监控的情况都不方便。

根据以上可知：Maxwell想做监控分析，选择row格式比较合适

配置mysql

sudo vim /etc/my.cnf

[mysqld]
datadir=/var/lib/mysql
socket=/var/lib/mysql/mysql.sock


symbolic-links=0


log-error=/var/log/mysqld.log
pid-file=/var/run/mysqld/mysqld.pid
# 唯一id，主从用
server-id=1
# binlog 日志的前缀
log-bin=mysql-bin
# binlog格式类型
binlog_format=row
# 监控的mysql数据库
binlog-do-db=mall

重启 MySQL 使配置生效

sudo systemctl restart mysqld

安装 Maxwell

解压

tar -zxvf /opt/software/maxwell-1.25.0.tar.gz -C /opt/module/

初始化Maxwell

• 在MySQL中建立一个maxwell库用于存储Maxwell的元数据

mysql -uroot -p123456

mysql> CREATE DATABASE maxwell ;

• 设置安全级别

mysql> set global validate_password_length=4;
mysql> set global validate_password_policy=0;

• 分配一个账号可以操作该数据库

mysql> GRANT ALL ON maxwell.* TO 'maxwell'@'%' IDENTIFIED BY '123456';

• 分配这个账号可以监控其他数据库的权限

mysql> GRANT SELECT ,REPLICATION SLAVE , REPLICATION CLIENT  ON *.* TO maxwell@'%';

配置 Maxwell

• 拷贝配置文件

cp config.properties.example config.properties

• 修改配置

# 生产方式 kafka
producer=kafka
# kafka broker 地址
kafka.bootstrap.servers=hadoop01:9092,hadoop02:9092,hadoop03:9092
# kafka topic
kafka_topic=ods_base_db
# 唯一id
clinet_id=maxwell_1
# maxwell元数据地址
host=hadoop01
port=3306


user=maxwell
password=123456

注意：默认还是输出到指定Kafka主题的一个kafka分区，因为多个分区并行可能会打乱binlog的顺序

如果要提高并行度，首先设置kafka的分区数>1,然后设置producer_partition_by属性

producer_partition_by=database|table|primary_key|random| column

• maxwell 启停脚本

#!/bin/bash
MAXWELL_HOME=/opt/module/maxwell-1.25.0
case $1 in
 "start")
   {
    $MAXWELL_HOME/bin/maxwell --config   $MAXWELL_HOME/config.properties >/dev/null 2>&1 &
  };;
  "stop")
  {
     ps -ef|grep Maxwell |grep -v grep|awk '{print $2}'|xargs kill >/dev/null 2>&1
  };;
   esac

测试

{"database":"mall","table":"student","type":"insert","ts":1615710386,"xid":248266,"commit":true,
"data":{"id":1,"name":"张三","age":100}}


{"database":"mall","table":"student","type":"insert","ts":1615710412,"xid":248298,"commit":true,
"data":{"id":2,"name":"李四","age":88}}

Maxwell 初始化数据

针对数据初始化的问题，Maxwell 提供了一个命令工具 maxwell-bootstrap 帮助我们完成数据初始化

maxwell-bootstrap 是基于 SELECT * FROM table 的方式进行全量数据初始化，不会产生多余的binlog！

bin/maxwell-bootstrap --user maxwell --password 123456 --host hadoop01 --database test --table test --client_id maxwell_1

其中client_id 是指另一个已启动的maxwell监控进程的client_id

注意：--bootstrapper=sync 时，在处理bootstrap时，会阻塞正常的binlog解析；--bootstrapper=async 时，不会阻塞。