MaxWell 使用笔记

MaxWell 使用笔记

目前使用场景：
采集MySQL 的Binlog日志文件，以JSON的形式发送的Kafka指定的分区中

Canal与MaxWell区别：
1、canal 每一条SQL会产生一条日志，如果该条Sql影响了多行数据，则已经会通过JSON_ARRAY的方式返回。
maxwell 以影响的数据为单位产生日志，即每影响一条数据就会产生一条日志。同一次操作xid相同
2、canal数据中会带入表结构。maxwell更简洁。

MaxWell使用总结

1、修改配置文件

vim /etc/my.cnf
在[mysqld]下插入log-error=/var/log/mysqld.log
pid-file=/var/run/mysqld/mysqld.pid
server-id= 1
log-bin=mysql-bin
binlog_format=row
binlog-do-db=test
#需要同步的业务数据库可以指定多个
binlog-do-db=test01
binlog-do-db=test02

2、重启MySQL

sudo systemctl restart mysqld

3、验证是否成功

cd /var/lib/mysql
查看初始文件，重启后应该会有一个新的 mysql-bin 文件生成 大小 154
修改/添加数据库信息
查看文件大小是否变化

4、初始化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 'max';
#分配账户监控全部数据库权限
mysql> GRANT  SELECT ,REPLICATION SLAVE , REPLICATION CLIENT  ON *.* TO maxwell@'%';

5、修改MaxWell配置文件

cp config.properties.example config.properties
vim config.properties
#添加如下内容
producer=kafka
kafka.bootstrap.servers=111.111.1.111:9092,222.222.2.222:9092,333.333.3.333:9092
#需要添加
kafka_topic=ods_db_maxwell
# mysql login info
host=111.111.1.111
user=maxwell
password=max
#需要添加 后续初始化会用
client_id=maxwell_1

6、注意事项
默认是输出到Kafka主题的一个分区下，因为多个分区并行操作可能会打乱binlog的顺序
因为是日志文件，所以没有严格的顺序要求，先消费哪条后消费哪条一样，可以如下修改来提高并行度

• 设置kafka分区数
• 设置maxwell配置文件中producer_partition_by属性

可选值producer_partition_by=database|table|primary_key| transaction_id | column
Database：一个数据库一个分区
Primary_kay:当前记录Binlog中，对主键取哈希值，均匀分部在分区中
transaction_id：事务id，按操作放入分区，如一次update修改了100条数据，这些全放在一个分区里 【JSON中的xid】
column：不使用主键了，使用表里的某一个字段

7、使用脚本启动
案例：

vim /home/111/bin/maxwell.sh
/opt/11/maxwell-1.25.0/bin/maxwell --config  /opt/11/maxwell-1.25.0/config.properties >/dev/null 2>&1 &

8、初始化MaxWell数据库之后，针对新增MySQL数据库操作

drop database maxwell

9、MaxWell bootstrap 功能

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

bootstrap不是将数据直接导入kafka或者hbase。是通过–client_id指定将数据交给哪个maxwell进程处理，在conf.properties中配置