解析mysql的binlog日志并推送到Kafka实时消费

采用Maxwell来实现实时解析mysql的binlog日志发送至kafka

1、开启mysql binlog 配置

my.cnf文件路径 /etc/my.cnf

[mysqld]
log-bin= /data/binlog/mysql-bin.log  #binlog位置及名称
log-bin-index=/data/binlog/mysql-bin.index  #自定义binlog路径时需要加上改选项
binlog-format="STATEMENT"  #binlog日志格式
expire_logs_days = 20      #binlog过期清理时间
max_binlog_size = 200M     #单个binlog日志文件大小
sync-binlog = 1            #开启binlog同步，将二进制日志同步到磁盘，保证数据完整性，在主从复制中也用到。
binlog-do-db=db1  # 需要记录binlog的数据库名，如果备份多个数据库，重复设置这个选项即可
binlog-ignore-db=db2  # 不需要记录binlog的数据库库名，如果备份多个数据库，重复设置这个选项即可
#最后两项可以用于主从复制场景

Maxwell简介

Maxwell是一个能实时读取MySQL二进制日志binlog，并生成 JSON 格式的消息，作为生产者发送给 Kafka，Kinesis、RabbitMQ、Redis、Google Cloud Pub/Sub、文件或其它平台的应用程序。它的常见应用场景有ETL、维护缓存、收集表级别的dml指标、增量到搜索引擎、数据分区迁移、切库binlog回滚方案等。官网(http://maxwells-daemon.io)、GitHub(https://github.com/zendesk/maxwell)

更多请见：Maxwell 详解

安装Maxwell

1、源代码:https://github.com/zendesk/maxwell
2、直接下载安装包:https://github.com/zendesk/maxwell/releases/download/v1.19.0/maxwell-1.19.0.tar.gz
3、解压

在kafka中创建一个topic

/opt/apps/kafka/bin/kafka-console-producer.sh --broker-list master:9092 --topic mysql_binlog

在后台启动mysql和maxwell

./bin/maxwell --user='root' --password='root' --port=3306 --host='192.168.227.180' --producer=kafka --kafka.bootstrap.servers=master:9092 --kafka_topic=mysql_binlog &

监控kafka中的数据

/opt/apps/kafka/bin/kafka-console-consumer.sh --zookeeper master:2181 --from-beginning --topic mysql_binlog

• 做insert时候的数据

{“database”:“localinfo”,“table”:“fang_house_price”,“type”:“insert”,“ts”:1540887971,“xid”:2574,“commit”:true,“data”:{“city_id”:“6D3323DA5A4763EFC1C06EA2DE21DFCE”,“statistics_date”:201810,“house_price”:58575.0}} {“database”:“localinfo”,“table”:“fang_house_price”,“type”:“insert”,“ts”:1540887971,“xid”:2588,“commit”:true,“data”:{“city_id”:“10D0597C95A533D4DD1FAF0CEE27700A”,“statistics_date”:201808,“house_price”:70638.0}} {“database”:“localinfo”,“table”:“fang_house_price”,“type”:“insert”,“ts”:1540887971,“xid”:2589,“commit”:true,“data”:{“city_id”:“10D0597C95A533D4DD1FAF0CEE27700A”,“statistics_date”:201809,“house_price”:70238.0}}

• 做delete的时候：

{“database”:“mtime_dw”,“table”:“flow_app_day”,“type”:“delete”,“ts”:1540888635,“xoffset”:369,“data”:{“device_id”:“c7d4d0a627e02e8LJ9SWSGUT8FUS4PJ”,“time”:“20170709144959”,“ip”:“223.104.108.158”,“log_type”:“2”,“event_type”:“0”,“type”:“app”,“dt”:20170709}} {“database”:“mtime_dw”,“table”:“flow_app_day”,“type”:“delete”,“ts”:1540888635,“commit”:true,“data”:{“device_id”:“c7d4d0a627e02e8LJ9SWSGUT8FUS4PJ”,“time”:“20170709145001”,“ip”:“223.104.108.158”,“log_type”:“2”,“event_type”:“0”,“type”:“app”,“dt”:20170709}}

• 做update操作的时候：

{“database”:“mtime_dw”,“table”:“flow_app_day”,“type”:“update”,“ts”:1540888780,“xoffset”:94,“data”:{“device_id”:“c8e9c2d5f4a0726bAHK0217111005110”,“time”:“20170709134113”,“ip”:“10.101.16.203”,“log_type”:“2”,“event_type”:“0”,“type”:“app”,“dt”:20170709},“old”:{“ip”:“36.5.96.145”}} {“database”:“mtime_dw”,“table”:“flow_app_day”,“type”:“update”,“ts”:1540888780,“xoffset”:95,“data”:{“device_id”:“c8e9c2d5f4a0726bAHK0217111005110”,“time”:“20170709134110”,“ip”:“10.101.16.203”,“log_type”:“2”,“event_type”:“0”,“type”:“app”,“dt”:20170709},“old”:{“ip”:“36.5.96.145”}} {“database”:“mtime_dw”,“table”:“flow_app_day”,“type”:“update”,“ts”:1540888780,“commit”:true,“data”:{“device_id”:“c8e9c2d5f4a0726bAHK0217111005110”,“time”:“20170709134106”,“ip”:“10.101.16.203”,“log_type”:“2”,“event_type”:“1”,“type”:“app”,“dt”:20170709},“old”:{“ip”:“36.5.96.145”}}

• 做drop和truncate操作的时候： kafka中未收到任何的消息记录。