MySQL系列：基於binlog的增量訂閱與消費（一）

　　在一些業務場景中，像在數據分析中我們有時候需要捕獲數據變化（CDC）；在數據審計中，我們也往往需要知道數據從這個點到另一個點的變化；同樣在實時分析中，我們有時候需要看到某個值得實時變化等。

要解決以上問題，我們可以實時解析mysql binlog日志，下面兩個工具可以很好的處理這個問題：

1. canal（阿里巴巴開源項目，純java開發）

2. python-mysql-replication（python開發）

 

使用場景：

1. MySQL到NoSQL的數據同步

2. MySQL到搜索引擎的復制

3. 當數據發生變化是清除數據緩存

4. 數據庫審計

5. 實時數據分析

 

本文主要說說python-mysql-replication。

 

介紹

python-mysql-replication是基於MySQL復制原理實現的，把自己偽裝成一個slave不斷的從MySQL數據庫獲取binlog並解析。

 當前版本(0.15)環境支持：

1. MySQL 5.5/5.6/5.7

2. Python >=2.6

3. Python 3.3/3.4/3.5/3.6（3.2不支持）

 

MySQL復制實現：

 

 

python-mysql-replication實現：

 

配置安裝

 安裝軟件

[root@mha-maxscale-1 ~]# pip install mysql-replication

MySQL授權

GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'replicator'@'192.168.3.%' IDENTIFIED BY '123456';

Binlog要滿足如下條件

MySQL>root@(none) 09:53:38>show variables like 'log_bin';
+---------------+-------+
| Variable_name | Value     |
+---------------+-------+
| log_bin       | ON       |
+---------------+-------+
1 row in set (0.01 sec)

MySQL>show variables like 'binlog_format';
+---------------+-------+
| Variable_name | Value     |
+---------------+-------+
| binlog_format | ROW      |
+---------------+-------+
1 row in set (0.00 sec)

MySQL>show variables like 'binlog_row_image';
+------------------+-------+
| Variable_name    | Value     |
+------------------+-------+
| binlog_row_image | FULL      |
+------------------+-------+
1 row in set (0.00 sec)

 

示例代碼：

[root@mha-maxscale-1 script]# cat mysql-replication.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-

from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
    DeleteRowsEvent,
    UpdateRowsEvent,
    WriteRowsEvent,
)
import sys
import json

def main():
    mysql_settings = {'host': '192.168.3.130',
                      'port': 3306, 'user': 'replicator', 'passwd': '123456'}
    stream = BinLogStreamReader(
        connection_settings=mysql_settings,
        server_id=101,
        blocking=True,
        only_schemas=['zow'],
        only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent],
        resume_stream=True,
        log_file='mysql-bin.000013', log_pos=6197)

    for binlogevent in stream:
        for row in binlogevent.rows:
            event = {"schema": binlogevent.schema, "table": binlogevent.table, "log_pos": binlogevent.packet.log_pos}
            if isinstance(binlogevent, DeleteRowsEvent):
                event["action"] = "delete"
                event["values"] = dict(row["values"].items())
                event = dict(event.items())
            elif isinstance(binlogevent, UpdateRowsEvent):
                event["action"] = "update"
                event["before_values"] = dict(row["before_values"].items())
                event["after_values"] = dict(row["after_values"].items())
                event = dict(event.items())
            elif isinstance(binlogevent, WriteRowsEvent):
                event["action"] = "insert"
                event["values"] = dict(row["values"].items())
                event = dict(event.items())
            print json.dumps(event)
            sys.stdout.flush()

    stream.close()


if __name__ == "__main__":
    main()

 

運行結果：

[root@mha-maxscale-1 script]# python mysql-replication.py 
{"action": "insert", "table": "t2", "log_pos": 4622, "values": {"tname": "hh", "id": 7}, "schema": "zow"}
{"log_pos": 4904, "after_values": {"tname": "ii", "id": 7}, "action": "update", "table": "t2", "before_values": {"tname": "hh", "id": 7}, "schema": "zow"}
{"log_pos": 4904, "after_values": {"tname": "ii", "id": 7}, "action": "update", "table": "t2", "before_values": {"tname": "hh", "id": 7}, "schema": "zow"}
{"action": "delete", "table": "t2", "log_pos": 5169, "values": {"tname": "ii", "id": 7}, "schema": "zow"}
{"action": "delete", "table": "t2", "log_pos": 5169, "values": {"tname": "ii", "id": 7}, "schema": "zow"}

 

更多例子見：https://github.com/noplay/python-mysql-replication/tree/master/examples