mysql biglog 的三种格式:
- 1)STATMENT模式:基于SQL语句的复制(statement-based replication, SBR),每一条会修改数据的sql语句会记录到binlog中
-
优点:binlog的日志比较少,减少了磁盘IO,提高性能
-
缺点:以下会导致master-slave中的数据不一致(如sleep()函数, last_insert_id(),以及 user-defined functions(udf)等会出现问题)
-
-
2)基于行的复制(row-based replication, RBR):记录最终数据的变化
-
优点:不存在 statement 模式的缺点。
-
缺点:会产生大量的日志,尤其是alter table的时候会让日志暴涨。
-
-
3)混合模式复制(mixed-based replication, MBR):以上两种模式的混合使用,一般的复制使 用STATEMENT模式保存binlog,对于STATEMENT模式无法复制的操作使用ROW模式保存binlog, MySQL会根据执行的SQL语句选择日志保存方式。
-
注:因为statement只有sql,没有数据,无法获 取原始的变更日志,所以一般建议为ROW模式
解析 biglog 多种方式之间的比较:
开启mysql的binlog功能并配置 maxwell:
- 1)添加mysql普通用户maxwell:
##进入mysql客户端,然后执行以下命令,进行授权
mysql -uroot -p
set global validate_password_policy=LOW;
set global validate_password_length=6;
CREATE USER 'maxwell'@'%' IDENTIFIED BY '123456';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE on *.* to 'maxwell'@'%';
flush privileges;
- 2)开启mysql的binlog机制:
## shell 进入一台节点编辑 mysql 的配置文件 my.cnf
sudo vim /etc/my.cnf
## 指定 biglog 格式 为row
log-bin= /var/lib/mysql/mysql-bin
binlog-format=ROW
server_id=1
## shell 重启mysql 服务
sudo service mysqld restart
## shell 验证
mysql -uroot -p
mysql> show variables like '%log_bin%';
- 3)安装max-well实现实时采集mysql数据:
-
下载max-well安装包,下载地址:
https://github.com/zendesk/maxwell/releases/download/v1.21.1/maxwell-1.21.1.tar.gz - 解压并修改配置文件:
-
## 编辑配置文件
cd xx/maxwell-1.21.1
cp config.properties.example config.properties
vim config.properties
## 指定数据 sink 为 kafka
producer=kafka
##根据实际的 kafka 配置
kafka.bootstrap.servers=node01:9092,node02:9092,node03:9092
host=node03.liz.com
port=3306
user=maxwell
password=123456
## 这个 topic 要手动创建
kafka_topic=maxwell_kafka
- 4)启动服务:
- 先后启动 zookeeper,kafka(安装步骤忽略)
- 创建 topic:kafka-topics.sh --create --topic maxwell_kafka --partitions 3 --replication-factor 2 --zookeeper node01:2181
- 启动 maxwell: cd xx/maxwell-1.21.1 bin/maxwell
- 5)插入数据并测试
- mysql 中建表:
-
CREATE DATABASE IF NOT EXISTS `test` DEFAULT CHARACTER SET utf8; USE `test`; /*Table structure for table `myuser` */ DROP TABLE IF EXISTS `myuser`; CREATE TABLE `myuser` ( `id` int(12) NOT NULL, `name` varchar(32) DEFAULT NULL, `age` varchar(32) DEFAULT NULL, PRIMARY KEY (`id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8; /*Data for the table `myuser` */ insert into `myuser`(`id`,`name`,`age`) values (1,'zhangsan',NULL),(2,'xxx',NULL),(3,'ggg',NULL),(5,'xxxx',NULL),(8,'skldjlskdf',NULL),(10,'ggggg',NULL),(99,'ttttt',NULL),(114,NULL,NULL),(121,'xxx',NULL);
-
- 启动 kafka 消费者线程消费 kafka 中的数据(--bootstrap-server 根据自己安装的 zookeeper 地址修改):kafka-console-consumer.sh --topic maxwell_kafka --from-beginning --bootstrap-server node01:9092,node02:9092,node03:9092
- mysql 中建表:
- 6)项目中 maxwell-kafka 配置样例(/xxx/maxwell-1.22.1/my_project.properties):
-
## my_project.properties log_level=INFO producer=kafka kafka.bootstrap.servers=node01:9092,node02:9092,node03:9092 ## maxwell 所在机器 host=yourhost.com user=maxwell password=123456 producer_ack_timeout = 600000 port=3306 ######### output format stuff ############### output_binlog_position=ture output_server_id=true output_thread_id=ture output_commit_info=true output_row_query=true output_ddl=false output_nulls=true output_xoffset=true output_schema_id=true ######### output format stuff ############### kafka_topic= yourtopic ## kafka recorde key 的生成方式,支持 array 和 hash kafka_key_format=hash kafka.compression.type=snappy kafka.retries=5 kafka.acks=all ## kafka 分区方式:表主键 producer_partition_by=primary_key kafka_partition_hash=murmur3 ############ kafka stuff ############# ## 在处理bootstrap时,是否会阻塞正常的binlog解析 async(异步)不会阻塞 ############## misc stuff ########### bootstrapper=async ############## misc stuff ########## ## 配置过滤:库名.表名, 下面的配置为:只采集db1 的 三张表 ## 支持正则过滤:include:db1:/table\\d{1}/ ############## filter ############### filter=exclude:*.*, include: db1.table1,include: db1.table2,include: db1.table3 ############## filter ###############
-
- 7) maxwell-kafka 启动脚本:
-
#!/bin/bash case $1 in "start" ){ nohup /xxx/maxwell-1.22.1/bin/maxwell --daemon --config /xxx/maxwell-1.22.1/my_project.properties 2>&1 >> /xxx/maxwell-1.22.1/maxwell.log & };; "stop"){ ps -ef | grep Maxwell | grep -v grep |awk '{print $2}' | xargs kill };; esac
-