基于binlog来分析mysql的行记录修改情况

昨晚写的脚本，因为个人python能力有限，本来想这不发这文，后来想想，没准会有哪位园友给出优化建议。

有些情况下，可能会统计在某个时间段内，MySQL修改了多少数据量？发生了多少事务？主要是哪些表格发生变动？变动的数量是怎么样的？ 但是却不需要行记录的修改内容，只需要了解 行数据的 变动情况。

这些情况部分可以通过监控来大致了解，但是也可以基于binlog来全盘分析，binlog的格式是row模式。

在写flashback的时候，顺带把这个也写了个脚步，使用python编写，都差不多原理，只是这个简单些，介于个人python弱的不行，性能可能还有很大的提升空间，也希望园友能协助优化下。

先贴python脚步的分析结果图如下，分为4个部分：事务耗时情况、事务影响行数情况、DML行数情况以及操作最频繁表格情况。

root@localhost:mysql3310.sock  14:42:29 [flashback]>show create table tbrow \G*************************** 1. row ***************************
       Table: tbrowCreate Table: CREATE TABLE `tbrow` (
  `auto_id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `sqltype` int(11) NOT NULL COMMENT &＃39;1 is insert,2 is update,3 is delete&＃39;,
  `tran_num` int(11) NOT NULL COMMENT &＃39;the transaction number&＃39;,
  `dbname` varchar(50) NOT NULL,
  `tbname` varchar(50) NOT NULL,  PRIMARY KEY (`auto_id`),  KEY `sqltype` (`sqltype`),  KEY `dbname` (`dbname`),  KEY `tbname` (`tbname`)
) ENGINE=InnoDB AUTO_INCREMENT=295151 DEFAULT CHARSET=utf81 row in set (0.00 sec)
 
root@localhost:mysql3310.sock  14:42:31 [flashback]>SHOW CREATE TABLE TBTRAN \G*************************** 1. row ***************************
       Table: TBTRANCreate Table: CREATE TABLE `tbtran` (
  `auto_id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `begin_time` datetime NOT NULL,
  `end_time` datetime NOT NULL,  PRIMARY KEY (`auto_id`)
) ENGINE=InnoDB AUTO_INCREMENT=6390 DEFAULT CHARSET=utf81 row in set (0.00 sec)

2.3 rowrecord

重点函数，分析binlog文件内容。这里有几个规律：

1. 每个事务的结束点，是以 &＃39;Xid = &＃39; 来查找

1. 事务的开始时间，是事务内的第一个 &＃39;Table_map&＃39; 行里边的时间

2. 事务的结束时间，是以 &＃39;Xid = &＃39;所在行的 里边的时间

2. 每个行数据是属于哪个表格，是以 &＃39;Table_map&＃39;来查找

3. DML的类型是按照 行记录开头的情况是否为：&＃39;### INSERT INTO&＃39; 、&＃39;### UPDATE&＃39; 、&＃39;### DELETE FROM&＃39;

4. 注意，单个事务可以包含多个表格多种DML多行数据修改的情况。

2.4 binlogdesc

描述分析结果，简单4个SQL分析。

1. 分析修改行数据的 事务耗时情况

2. 分析修改行数据的 事务影响行数情况

3. 分析DML分布情况

4. 分析 最多DML操作的表格 ，取前十个分析

2.5 closeconn

关闭数据库连接。

首先，确保python安装了pymysql模块，把python脚本拷贝到文件 queryanalyse.py。

然后，把要分析的binlog文件先用 mysqlbinlog 指令分析存储，具体binlog的文件说明，可以查看之前的博文：关于binary log那些事——认真码了好长一篇。mysqlbinlog的指令使用方法，可以详细查看文档：https://dev.mysql.com/doc/refman/5.7/en/mysqlbinlog.html 。

比较常用通过指定开始时间跟结束时间来分析 binlog文件。

mysqlbinlog --start-datetime=&＃39;2017-04-23 00:00:03&＃39; --stop-datetime=&＃39;2017-04-23 00:30:00&＃39; --base64-output=decode-rows -v /data/mysql/logs/mysql-bin.007335 > /tmp/binlog_test.log

分析后，可以把这个 binlog_test.log文件拷贝到其他空闲服务器执行分析，只需要有个空闲的DB来存储分析记录即可。

假设这个时候，拷贝 binlog_test.log到测试服务器上，测试服务器上的数据库可以用来存储分析内容，则可以执行python脚本了，注意要进入到python脚本的目录中，或者指定python脚本路径。

python queryanalyse.py -h=127.0.0.1 -P=3310 -u=root -p=password -f= /tmp/binlog_test.log -tt=flashback.tbtran -tr=flashback.tbrow

没了，就等待输出吧。

性能是硬伤，在虚拟机上测试，大概500M的binlog文件需要分析2-3min，有待提高！

  1 import pymysql  2 from pymysql.cursors import DictCursor  3 import re  4 import os  5 import sys  6 import datetime  7 import time  8 import logging  9 import importlib 10 importlib.reload(logging) 11 logging.basicConfig(level=logging.DEBUG,format=&＃39;%(asctime)s %(levelname)s %(message)s &＃39;) 12  13  14 usage=&＃39;&＃39;&＃39; usage: python [script&＃39;s path] [option] 15 ALL options need to assign: 16  17 -h     : host, the database host，which database will store the results after analysis 
 18 -u     : user, the db user 19 -p     : password, the db user&＃39;s password 20 -P     : port, the db port 21 -f     : file path, the binlog file 22 -tr    : table name for record , the table name to store the row record 23 -tt    : table name for transaction, the table name to store transactions 24 Example: python queryanalyse.py -h=127.0.0.1 -P=3310 -u=root -p=password -f=/tmp/stock_binlog.log -tt=flashback.tbtran -tr=flashback.tbrow 25  26 &＃39;&＃39;&＃39; 27  28 class queryanalyse: 29     def init(self): 30         #初始化 31         self.host=&＃39;&＃39; 32         self.user=&＃39;&＃39; 33         self.password=&＃39;&＃39; 34         self.port=&＃39;3306&＃39; 35         self.fpath=&＃39;&＃39; 36         self.tbrow=&＃39;&＃39; 37         self.tbtran=&＃39;&＃39; 38  39         self._get_db() 40         logging.info(&＃39;assign values to parameters is done:host={},user={},password=***,port={},fpath={},tb_for_record={},tb_for_tran={}&＃39;.format(self.host,self.user,self.port,self.fpath,self.tbrow,self.tbtran)) 41  42         self.mysqlcOnn= pymysql.connect(host=self.host, user=self.user, password=self.password, port=self.port,charset=&＃39;utf8&＃39;) 43         self.cur = self.mysqlconn.cursor(cursor=DictCursor) 44         logging.info(&＃39;MySQL which userd to store binlog event connection is ok&＃39;) 45  46         self.begin_time=&＃39;&＃39; 47         self.end_time=&＃39;&＃39; 48         self.db_name=&＃39;&＃39; 49         self.tb_name=&＃39;&＃39; 50  51     def _get_db(self): 52         #解析用户输入的选项参数值，这里对password的处理是明文输入，可以自行处理成是input格式， 53         #由于可以拷贝binlog文件到非线上环境分析，所以password这块，没有特殊处理 54         logging.info(&＃39;begin to assign values to parameters&＃39;) 55         if len(sys.argv) == 1: 56             print(usage) 57             sys.exit(1) 58         elif sys.argv[1] == &＃39;--help&＃39;: 59             print(usage) 60             sys.exit() 61         elif len(sys.argv) > 2: 62             for i in sys.argv[1:]: 63                 _argv = i.split(&＃39;=&＃39;) 64                 if _argv[0] == &＃39;-h&＃39;: 65                     self.host = _argv[1] 66                 elif _argv[0] == &＃39;-u&＃39;: 67                     self.user = _argv[1] 68                 elif _argv[0] == &＃39;-P&＃39;: 69                     self.port = int(_argv[1]) 70                 elif _argv[0] == &＃39;-f&＃39;: 71                     self.fpath = _argv[1] 72                 elif _argv[0] == &＃39;-tr&＃39;: 73                     self.tbrow = _argv[1] 74                 elif _argv[0] == &＃39;-tt&＃39;: 75                     self.tbtran = _argv[1] 76                 elif _argv[0] == &＃39;-p&＃39;: 77                     self.password = _argv[1] 78                 else: 79                     print(usage) 80  81     def create_tab(self): 82         #创建两个表格：一个用户存储事务情况，一个用户存储每一行数据修改的情况 83         #注意，一个事务可以存储多行数据修改的情况 84         logging.info(&＃39;creating table ...&＃39;) 85         create_tb_sql =&＃39;&＃39;&＃39;CREATE TABLE IF NOT EXISTS  {} ( 86                           `auto_id` int(10) unsigned NOT NULL AUTO_INCREMENT, 87                           `begin_time` datetime NOT NULL, 88                           `end_time` datetime NOT NULL, 89                           PRIMARY KEY (`auto_id`) 90                         ) ENGINE=InnoDB DEFAULT CHARSET=utf8; 91                         CREATE TABLE IF NOT EXISTS  {} ( 92                           `auto_id` int(10) unsigned NOT NULL AUTO_INCREMENT, 93                           `sqltype` int(11) NOT NULL COMMENT &＃39;1 is insert,2 is update,3 is delete&＃39;, 94                           `tran_num` int(11) NOT NULL COMMENT &＃39;the transaction number&＃39;, 95                           `dbname` varchar(50) NOT NULL, 96                           `tbname` varchar(50) NOT NULL, 97                           PRIMARY KEY (`auto_id`), 98                           KEY `sqltype` (`sqltype`), 99                           KEY `dbname` (`dbname`),100                           KEY `tbname` (`tbname`)101                         ) ENGINE=InnoDB DEFAULT CHARSET=utf8;102                         truncate table {};103                         truncate table {};104                         &＃39;&＃39;&＃39;.format(self.tbtran,self.tbrow,self.tbtran,self.tbrow)105 106         self.cur.execute(create_tb_sql)107         logging.info(&＃39;created table {} and {}&＃39;.format(self.tbrow,self.tbtran))108 109     def rowrecord(self):110         #处理每一行binlog111         #事务的结束采用 &＃39;Xid =&＃39; 来划分112         #分析结果，按照一个事务为单位存储提交一次到db113         try:114             tran_num=1    #事务数115             record_sql=&＃39;&＃39; #行记录的insert sql116             tran_sql=&＃39;&＃39;   #事务的insert sql117 118             self.create_tab()119 120             with open(self.fpath,&＃39;r&＃39;) as binlog_file:121                 logging.info(&＃39;begining to analyze the binlog file ,this may be take a long time !!!&＃39;)122                 logging.info(&＃39;analyzing...&＃39;)123 124                 for bline in binlog_file:125 126                     if bline.find(&＃39;Table_map:&＃39;) != -1:127                         l = bline.index(&＃39;server&＃39;)128                         n = bline.index(&＃39;Table_map&＃39;)129                         begin_time = bline[:l:].rstrip(&＃39; &＃39;).replace(&＃39;#&＃39;, &＃39;20&＃39;)130 131                         if record_sql==&＃39;&＃39;:132                             self.begin_time = begin_time[0:4] + &＃39;-&＃39; + begin_time[4:6] + &＃39;-&＃39; + begin_time[6:]133 134                         self.db_name = bline[n::].split(&＃39; &＃39;)[1].replace(&＃39;`&＃39;, &＃39;&＃39;).split(&＃39;.&＃39;)[0]135                         self.tb_name = bline[n::].split(&＃39; &＃39;)[1].replace(&＃39;`&＃39;, &＃39;&＃39;).split(&＃39;.&＃39;)[1]136                         bline=&＃39;&＃39;137 138                     elif bline.startswith(&＃39;### INSERT INTO&＃39;):139                        record_sql=record_sql+"insert into {}(sqltype,tran_num,dbname,tbname) VALUES (1,{},&＃39;{}&＃39;,&＃39;{}&＃39;);".format(self.tbrow,tran_num,self.db_name,self.tb_name)140 141                     elif bline.startswith(&＃39;### UPDATE&＃39;):142                        record_sql=record_sql+"insert into {}(sqltype,tran_num,dbname,tbname) VALUES (2,{},&＃39;{}&＃39;,&＃39;{}&＃39;);".format(self.tbrow,tran_num,self.db_name,self.tb_name)143 144                     elif bline.startswith(&＃39;### DELETE FROM&＃39;):145                        record_sql=record_sql+"insert into {}(sqltype,tran_num,dbname,tbname) VALUES (3,{},&＃39;{}&＃39;,&＃39;{}&＃39;);".format(self.tbrow,tran_num,self.db_name,self.tb_name)146 147                     elif bline.find(&＃39;Xid =&＃39;) != -1:148 149                         l = bline.index(&＃39;server&＃39;)150                         end_time = bline[:l:].rstrip(&＃39; &＃39;).replace(&＃39;#&＃39;, &＃39;20&＃39;)151                         self.end_time = end_time[0:4] + &＃39;-&＃39; + end_time[4:6] + &＃39;-&＃39; + end_time[6:]152                         tran_sql=record_sql+"insert into {}(begin_time,end_time) VALUES (&＃39;{}&＃39;,&＃39;{}&＃39;)".format(self.tbtran,self.begin_time,self.end_time)153 154                         self.cur.execute(tran_sql)155                         self.mysqlconn.commit()156                         record_sql = &＃39;&＃39;157                         tran_num += 1158 159         except Exception:160             return &＃39;funtion rowrecord error&＃39;161 162     def binlogdesc(self):163         sql=&＃39;&＃39;164         t_num=0165         r_num=0166         logging.info(&＃39;Analysed result printing...\n&＃39;)167         #分析总的事务数跟行修改数量168         sql="select &＃39;tbtran&＃39; name,count(*) nums from {}  union all select &＃39;tbrow&＃39; name,count(*) nums from {};".format(self.tbtran,self.tbrow)169         self.cur.execute(sql)170         rows=self.cur.fetchall()171         for row in rows:172             if row[&＃39;name&＃39;]==&＃39;tbtran&＃39;:173                 t_num = row[&＃39;nums&＃39;]174             else:175                 r_num = row[&＃39;nums&＃39;]176         print(&＃39;This binlog file has {} transactions, {} rows are changed &＃39;.format(t_num,r_num))177 178         # 计算 最耗时 的单个事务179         # 分析每个事务的耗时情况,分为5个时间段来描述180         # 这里正常应该是 以毫秒来分析的，但是binlog中，只精确时间到second181         sql=&＃39;&＃39;&＃39;select 
182                       count(case when cost_sec between 0 and 1 then 1 end ) cos_1,183                       count(case when cost_sec between 1.1 and 5 then 1 end ) cos_5,184                       count(case when cost_sec between 5.1 and 10 then 1 end ) cos_10,185                       count(case when cost_sec between 10.1 and 30 then 1 end ) cos_30,186                       count(case when cost_sec >30.1 then 1 end ) cos_more,187                       max(cost_sec) cos_max188                 from 
189                 (190                         select 
191                             auto_id,timestampdiff(second,begin_time,end_time) cost_sec192                         from {}193                 ) a;&＃39;&＃39;&＃39;.format(self.tbtran)194         self.cur.execute(sql)195         rows=self.cur.fetchall()196 197         for row in rows:198             print(&＃39;The most cost time : {} &＃39;.format(row[&＃39;cos_max&＃39;]))199             print(&＃39;The distribution map of each transaction costed time: &＃39;)200             print(&＃39;Cost time between    0 and  1 second : {} , {}%&＃39;.format(row[&＃39;cos_1&＃39;],int(row[&＃39;cos_1&＃39;]*100/t_num)))201             print(&＃39;Cost time between  1.1 and  5 second : {} , {}%&＃39;.format(row[&＃39;cos_5&＃39;], int(row[&＃39;cos_5&＃39;] * 100 / t_num)))202             print(&＃39;Cost time between  5.1 and 10 second : {} , {}%&＃39;.format(row[&＃39;cos_10&＃39;], int(row[&＃39;cos_10&＃39;] * 100 / t_num)))203             print(&＃39;Cost time between 10.1 and 30 second : {} , {}%&＃39;.format(row[&＃39;cos_30&＃39;], int(row[&＃39;cos_30&＃39;] * 100 / t_num)))204             print(&＃39;Cost time                     > 30.1 : {} , {}%\n&＃39;.format(row[&＃39;cos_more&＃39;], int(row[&＃39;cos_more&＃39;] * 100 / t_num)))205 206         # 计算 单个事务影响行数最多 的行数量207         # 分析每个事务 影响行数 情况,分为5个梯度来描述208         sql=&＃39;&＃39;&＃39;select 
209                     count(case when nums between 0 and 10 then 1 end ) row_1,210                     count(case when nums between 11 and 100 then 1 end ) row_2,211                     count(case when nums between 101 and 1000 then 1 end ) row_3,212                     count(case when nums between 1001 and 10000 then 1 end ) row_4,213                     count(case when nums >10001 then 1 end ) row_5,214                     max(nums) row_max215                from 
216                   (217                     select 
218                              count(*) nums219                     from {} group by tran_num220                    ) a;&＃39;&＃39;&＃39;.format(self.tbrow)221         self.cur.execute(sql)222         rows=self.cur.fetchall()223 224         for row in rows:225             print(&＃39;The most changed rows for each row: {} &＃39;.format(row[&＃39;row_max&＃39;]))226             print(&＃39;The distribution map of each transaction changed rows : &＃39;)227             print(&＃39;Changed rows between    1 and    10 second : {} , {}%&＃39;.format(row[&＃39;row_1&＃39;],int(row[&＃39;row_1&＃39;]*100/t_num)))228             print(&＃39;Changed rows between   11 and   100 second : {} , {}%&＃39;.format(row[&＃39;row_2&＃39;], int(row[&＃39;row_2&＃39;] * 100 / t_num)))229             print(&＃39;Changed rows between  101 and  1000 second : {} , {}%&＃39;.format(row[&＃39;row_3&＃39;], int(row[&＃39;row_3&＃39;] * 100 / t_num)))230             print(&＃39;Changed rows between 1001 and 10000 second : {} , {}%&＃39;.format(row[&＃39;row_4&＃39;], int(row[&＃39;row_4&＃39;] * 100 / t_num)))231             print(&＃39;Changed rows                       > 10001 : {} , {}%\n&＃39;.format(row[&＃39;row_5&＃39;], int(row[&＃39;row_5&＃39;] * 100 / t_num)))232 233         # 分析 各个行数 DML的类型情况234         # 描述 delete，insert，update的分布情况235         sql=&＃39;select sqltype ,count(*) nums from {} group by sqltype ;&＃39;.format(self.tbrow)236         self.cur.execute(sql)237         rows=self.cur.fetchall()238 239         print(&＃39;The distribution map of the {} changed rows : &＃39;.format(r_num))240         for row in rows:241 242             if row[&＃39;sqltype&＃39;]==1:243                 print(&＃39;INSERT rows :{} , {}% &＃39;.format(row[&＃39;nums&＃39;],int(row[&＃39;nums&＃39;]*100/r_num)))244             if row[&＃39;sqltype&＃39;]==2:245                 print(&＃39;UPDATE rows :{} , {}% &＃39;.format(row[&＃39;nums&＃39;],int(row[&＃39;nums&＃39;]*100/r_num)))246             if row[&＃39;sqltype&＃39;]==3:247                 print(&＃39;DELETE rows :{} , {}%\n &＃39;.format(row[&＃39;nums&＃39;],int(row[&＃39;nums&＃39;]*100/r_num)))248 249         # 描述 影响行数 最多的表格250         # 可以分析是哪些表格频繁操作，这里显示前10个table name251         sql = &＃39;&＃39;&＃39;select 
252                       dbname,tbname ,253                       count(*) ALL_rows,254                       count(*)*100/{} per,255                       count(case when sqltype=1 then 1 end) INSERT_rows,256                       count(case when sqltype=2 then 1 end) UPDATE_rows,257                       count(case when sqltype=3 then 1 end) DELETE_rows258                 from {} 
259                 group by dbname,tbname 
260                 order by ALL_rows desc 
261                 limit 10;&＃39;&＃39;&＃39;.format(r_num,self.tbrow)262         self.cur.execute(sql)263         rows = self.cur.fetchall()264 265         print(&＃39;The distribution map of the {} changed rows : &＃39;.format(r_num))266         print(&＃39;tablename&＃39;.ljust(50),267               &＃39;|&＃39;,&＃39;changed_rows&＃39;.center(15),268               &＃39;|&＃39;,&＃39;percent&＃39;.center(10),269               &＃39;|&＃39;,&＃39;insert_rows&＃39;.center(18),270               &＃39;|&＃39;,&＃39;update_rows&＃39;.center(18),271               &＃39;|&＃39;,&＃39;delete_rows&＃39;.center(18)272               )273         print(&＃39;-------------------------------------------------------------------------------------------------------------------------------------------------&＃39;)274         for row in rows:275             print((row[&＃39;dbname&＃39;]+&＃39;.&＃39;+row[&＃39;tbname&＃39;]).ljust(50),276                   &＃39;|&＃39;,str(row[&＃39;ALL_rows&＃39;]).rjust(15),277                   &＃39;|&＃39;,(str(int(row[&＃39;per&＃39;]))+&＃39;%&＃39;).rjust(10),278                   &＃39;|&＃39;,str(row[&＃39;INSERT_rows&＃39;]).rjust(10)+&＃39; , &＃39;+(str(int(row[&＃39;INSERT_rows&＃39;]*100/row[&＃39;ALL_rows&＃39;]))+&＃39;%&＃39;).ljust(5),279                   &＃39;|&＃39;,str(row[&＃39;UPDATE_rows&＃39;]).rjust(10)+&＃39; , &＃39;+(str(int(row[&＃39;UPDATE_rows&＃39;]*100/row[&＃39;ALL_rows&＃39;]))+&＃39;%&＃39;).ljust(5),280                   &＃39;|&＃39;,str(row[&＃39;DELETE_rows&＃39;]).rjust(10)+&＃39; , &＃39;+(str(int(row[&＃39;DELETE_rows&＃39;]*100/row[&＃39;ALL_rows&＃39;]))+&＃39;%&＃39;).ljust(5),281                   )282         print(&＃39;\n&＃39;)283 284         logging.info(&＃39;Finished to analyse the binlog file !!!&＃39;)285 286     def closeconn(self):287         self.cur.close()288         logging.info(&＃39;release db connections\n&＃39;)289 290 def main():291     p = queryanalyse()292     p.rowrecord()293     p.binlogdesc()294     p.closeconn()295 296 if name == "main":297     main()

以上就是基于binlog来分析mysql的行记录修改情况的详细内容，更多请关注 第一PHP社区 其它相关文章！