解析mysql binlog

目前有个功能需要将mysql中的数据同步到数仓（hdfs，kudu或者其他存储）中去，如果离线的方式可以考虑sqoop，spark，缺点就是延时比较高，可能一个小时，或者一天；数据一致性很难得到保证，mysql中的数据经常会变（当然你可以设置一个周期，例如按天同步的话，每次同步都同步过去一个月或者更长周期的数据）

现在的需求是需要数仓和mysql中的数据保持在秒级别的一致；目前的思路就是实时监听mysql binlog日志，使mysql中的数据变化实时同步到数仓中（这里需要注意的是，我们只监听insert，update，delete这几种event）

监听mysql binlog工具有很多，目前大多数使用

mysql-binlog-connector-java （https://github.com/shyiko/mysql-binlog-connector-java）
canal (阿里巴巴开源)

canal集群方式部署，系统比较臃肿，复杂性也相对较高，可定制性也比较差，所以我们没有考虑这种方案，采用了mysql-binlog-connector-java

首先需要修改mysql配置文件my.cnf，然后重启

server-id               = 1
log_bin                 = /var/log/mysql/mysql-bin.log
expire_logs_days        = 10
max_binlog_size   = 100M
binlog_format = ROW
#bind-address           = 127.0.0.1

重启后状态如下

查看状态
mysql> show variables like '%log_bin%';
+---------------------------------+--------------------------------+
| Variable_name                   | Value                          |
+---------------------------------+--------------------------------+
| log_bin                         | ON                             |
| log_bin_basename                | /var/log/mysql/mysql-bin       |
| log_bin_index                   | /var/log/mysql/mysql-bin.index |
| log_bin_trust_function_creators | OFF                            |
| log_bin_use_v1_row_events       | OFF                            |
| sql_log_bin                     | ON                             |
+---------------------------------+--------------------------------+
6 rows in set (0.00 sec)

接下分析binlog，代码如下

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.Event;

public class BinlogParse {

    public static void main(String[] args) throws Exception {
        final BinaryLogClient client = new BinaryLogClient("10.23.92.189", 3306, "root", "hadoop");
        client.setBinlogFilename("mysql-bin.000002");
        client.setBinlogPosition(123);
        client.registerEventListener(new BinaryLogClient.EventListener() {

            public void onEvent(Event event) {

                System.out.println(event.toString());
                System.out.println(client.getBinlogPosition());

            }
        });
        client.connect();
    }
}

这里需要注意的是，如果你不指定如下参数，程序将从mysql当前binlog位置开始同步数据，这显然不是我们需要的，更多的时候我们需要灵活的从任意位置读取数据

client.setBinlogFilename("mysql-bin.000002");//可以随意设置，但不能为空
client.setBinlogPosition(123);//从指定的位置读取binlog，具体位置可以在mysql中查看

读出来的binlog日志我们会封装成一个json字符串push到kafka，然后再从kafka消费数据到数仓(我们这边采用的是kudu)，这样的话数仓中就有最实时的数据，olap引擎采用impala

需要注意的是delete from table 这种操作，表中数据是一条一条删除，如果表中数据非常大的话binlog日志也会非常大

猜你喜欢