在一些業務場景中，像在數據分析中我們有時候需要捕獲數據變化（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授權

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

Binlog要滿足如下條件

 1 MySQL>root@(none) 09:53:38>show variables like ‘log_bin‘;
 2 +---------------+-------+
 3 | Variable_name | Value     |
 4 +---------------+-------+
 5 | log_bin       |
 
 ON       |
 6 +---------------+-------+
 7 1 row in set (0.01 sec)
 8 
 9 MySQL>show variables like ‘binlog_format‘;
10 +---------------+-------+
11 | Variable_name | Value     |
12 +---------------+-------+
13 | binlog_format | ROW      |
14 +---------------+-------+
15 1 row in set (0.00 sec)
16 
17 MySQL>show variables like ‘binlog_row_image‘;
18 +------------------+-------+
19 | Variable_name    | Value     |
20 +------------------+-------+
21 | binlog_row_image | FULL      |
22 +------------------+-------+
23 1 row in set (0.00 sec)

示例代碼：

 1 [root@mha-maxscale-1 script]# cat mysql-replication.py
 2 #!/usr/bin/env python
 3 # -*- coding: utf-8 -*-
 4 
 5 from pymysqlreplication import BinLogStreamReader
 6 from pymysqlreplication.row_event import (
 7     DeleteRowsEvent,
 8     UpdateRowsEvent,
 9     WriteRowsEvent,
10 )
11 import sys
12 import json
13 
14 def main():
15     mysql_settings = {‘host‘: ‘192.168.3.130‘,
16                       ‘port‘: 3306, ‘user‘: ‘replicator‘, ‘passwd‘: ‘123456‘}
17     stream = BinLogStreamReader(
18         connection_settings=mysql_settings,
19         server_id=101,
20         blocking=True,
21         only_schemas=[‘zow‘],
22         only_events=[DeleteRowsEvent, WriteRowsEvent, UpdateRowsEvent],
23         resume_stream=True,
24         log_file=‘mysql-bin.000013‘, log_pos=6197)
25 
26     for binlogevent in stream:
27         for row in binlogevent.rows:
28             event = {"schema": binlogevent.schema, "table": binlogevent.table, "log_pos": binlogevent.packet.log_pos}
29             if isinstance(binlogevent, DeleteRowsEvent):
30                 event["action"] = "delete"
31                 event["values"] = dict(row["values"].items())
33                 event = dict(event.items())
34             elif isinstance(binlogevent, UpdateRowsEvent):
35                 event["action"] = "update"
36                 event["before_values"] = dict(row["before_values"].items())
37                 event["after_values"] = dict(row["after_values"].items())
38                 event = dict(event.items())
39             elif isinstance(binlogevent, WriteRowsEvent):
40                 event["action"] = "insert"
41                 event["values"] = dict(row["values"].items())
42                 event = dict(event.items())
43             print json.dumps(event)
44             sys.stdout.flush()
45 
46     stream.close()
47 
48 
49 if __name__ == "__main__":
50     main()

運行結果：

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

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

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