一、先上效果图，再做阐述

DEMO

主从延迟，快速定位哪个表引起的

二、背景

当生产业务出现了主从延迟，DBA无法立刻知道是什么操作引起的，需要解析binlog/relay log、分析，处理问题的时间就是业务被影响的时间，所以不够快。

每张表或者每个库的DML的历史趋势，DBA是需要了解TOP N的，才能更好的分析业务、帮助业务。

其他，不一一罗列，活学活用，方可对症下药

三、实现工具

binlog2sql工具(本质python脚本)进行微改

MySQL

Grafana

四、实现原理

原理比较简单：

binlog2sql，模拟slave，与生产master建立主从关系(类似canal)。

解析binlog event，分析db、table_name、dml类型(insert/update/delete)、event时间。

按照分钟级别、小时级别对db、table_name、dml类型、dml总数、每分钟/每小时记录。

为什么选择生产master，而不是生产slave节点建里关系？因为如果生产slave发生主从延迟，那么binlog2sql依赖延迟的slave，实时性会不可控。

一定要记录binlog event时间作为dml发生时间，记录入库也要记录当前时间，这样可以准确判断DML发生时间同时，也可以判断Binlog2SQL的延迟时间。

这里记录结果用的MySQL并按照分钟、小时两个维度，为了利用MySQL减少对结果集的聚合统计，比较通用。可以选择时序DB或者ClickHouse，效果更佳。

五、源码附上

记录结果MySQL表结构

CREATE TABLE `t_monitor_mysql_table_dml_by_minute` (

`id` bigint(20) NOT NULL AUTO_INCREMENT,

`ip` varchar(100) DEFAULT NULL,

`port` int(11) DEFAULT NULL,

`table_schema` varchar(100) DEFAULT NULL,

`table_name` varchar(200) DEFAULT NULL,

`dml_type` varchar(100) DEFAULT NULL,

`dml_count` bigint(20) DEFAULT NULL,

`start_time` datetime DEFAULT NULL,

`create_time` datetime DEFAULT NULL,

PRIMARY KEY (`id`),

UNIQUE KEY `ix_unique_1` (`ip`,`port`,`start_time`,`table_schema`,`table_name`,`dml_type`)

) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

CREATE TABLE `t_monitor_mysql_table_dml_by_hour` (

`id` bigint(20) NOT NULL AUTO_INCREMENT,

`ip` varchar(100) DEFAULT NULL,

`port` int(11) DEFAULT NULL,

`table_schema` varchar(100) DEFAULT NULL,

`table_name` varchar(200) DEFAULT NULL,

`dml_type` varchar(100) DEFAULT NULL,

`dml_count` bigint(20) DEFAULT NULL,

`start_time` datetime DEFAULT NULL,

`create_time` datetime DEFAULT NULL,

PRIMARY KEY (`id`),

UNIQUE KEY `ix_unique_1` (`ip`,`port`,`start_time`,`table_schema`,`table_name`,`dml_type`),

KEY `ix_1` (`ip`,`port`,`table_schema`,`table_name`,`dml_type`,`start_time`)

) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4;

为什么要建立唯一索引呢？

因为累积计数，利用了insert into ... on duplicate key update dml_count=dml_count+values(dml_count);

改写binlog2sql脚本 ：dbms_monitor_table_dml_stat.py

#!/usr/bin/env python

# -*- coding: utf-8 -*-

import sys

import datetime

import pymysql

from pymysqlreplication import BinLogStreamReader

from pymysqlreplication.event import QueryEvent, RotateEvent, FormatDescriptionEvent

from binlog2sql_util import command_line_args, concat_sql_from_binlog_event, create_unique_file, temp_open, \

reversed_lines, is_dml_event, event_type

from db_help import *

class Binlog2sql(object):

def __init__(self, connection_settings, start_file=None, start_pos=None, end_file=None, end_pos=None,

start_time=None, stop_time=None, only_schemas=None, only_tables=None, no_pk=False,

flashback=False, stop_never=False, back_interval=1.0, only_dml=True, sql_type=None):

"""

conn_setting: {'host': 127.0.0.1, 'port': 3306, 'user': user, 'passwd': passwd, 'charset': 'utf8'}

"""

if not start_file:

raise ValueError('Lack of parameter: start_file')

self.conn_setting = connection_settings

self.start_file = start_file

self.start_pos = start_pos if start_pos else 4 # use binlog v4

self.end_file = end_file if end_file else start_file

self.end_pos = end_pos

if start_time:

self.start_time = datetime.datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")

else:

self.start_time = datetime.datetime.strptime('1980-01-01 00:00:00', "%Y-%m-%d %H:%M:%S")

if stop_time:

self.stop_time = datetime.datetime.strptime(stop_time, "%Y-%m-%d %H:%M:%S")

else:

self.stop_time = datetime.datetime.strptime('2999-12-31 00:00:00', "%Y-%m-%d %H:%M:%S")

self.only_schemas = only_schemas if only_schemas else None

self.only_tables = only_tables if only_tables else None

self.no_pk, self.flashback, self.stop_never, self.back_interval = (no_pk, flashback, stop_never, back_interval)

self.only_dml = only_dml

self.sql_type = [t.upper() for t in sql_type] if sql_type else []

self.binlogList = []

self.connection = pymysql.connect(**self.conn_setting)

with self.connection as cursor:

cursor.execute("SHOW MASTER STATUS")

self.eof_file, self.eof_pos = cursor.fetchone()[:2]

cursor.execute("SHOW MASTER LOGS")

bin_index = [row[0] for row in cursor.fetchall()]

if self.start_file not in bin_index:

raise ValueError('parameter error: start_file %s not in mysql server' % self.start_file)

binlog2i = lambda x: x.split('.')[1]

for binary in bin_index:

if binlog2i(self.start_file) <= binlog2i(binary) <= binlog2i(self.end_file):

self.binlogList.append(binary)

cursor.execute("SELECT @@server_id")

self.server_id = cursor.fetchone()[0]

if not self.server_id:

raise ValueError('missing server_id in %s:%s' % (self.conn_setting['host'], self.conn_setting['port']))

def process_binlog(self,ip,port):

ms = MYSQL('记录结果集MySQL IP', 3306, "DB", 1, 30) # 需要人工更改，记录结果使用长连接，提高效率

con = ms.GetConnect()

stream = BinLogStreamReader(connection_settings=self.conn_setting, server_id=self.server_id,

log_file=self.start_file, log_pos=self.start_pos, only_schemas=self.only_schemas,

only_tables=self.only_tables, resume_stream=True, blocking=True)

flag_last_event = False

e_start_pos, last_pos = stream.log_pos, stream.log_pos

# to simplify code, we do not use flock for tmp_file.

tmp_file = create_unique_file('%s.%s' % (self.conn_setting['host'], self.conn_setting['port']))

for binlog_event in stream:

if (is_dml_event(binlog_event) and event_type(binlog_event) in self.sql_type):

event_time = datetime.datetime.fromtimestamp(binlog_event.timestamp)

event_time = str(event_time.strftime('%Y-%m-%d %H:%M:%S'))

sql ="insert into t_monitor_mysql_table_dml_by_minute(ip,port,table_schema,table_name,dml_type,start_time,create_time,dml_count) values('{0}',{1},'{2}','{3}','{4}',date_format('{5}','%Y-%m-%d %H:%i:00'),now(),1) on duplicate key update dml_count=dml_count+values(dml_count);".format(ip,port,binlog_event.schema,binlog_event.table,event_type(binlog_event),event_time)

ms.ExecNonQuery(sql, con)

sql = "insert into t_monitor_mysql_table_dml_by_hour(ip,port,table_schema,table_name,dml_type,start_time,create_time,dml_count) values('{0}',{1},'{2}','{3}','{4}',date_format('{5}','%Y-%m-%d %H:00:00'),now(),1) on duplicate key update dml_count=dml_count+values(dml_count);".format(ip, port, binlog_event.schema, binlog_event.table, event_type(binlog_event), event_time)

ms.ExecNonQuery(sql, con)

if not (isinstance(binlog_event, RotateEvent) or isinstance(binlog_event, FormatDescriptionEvent)):

last_pos = binlog_event.packet.log_pos

if flag_last_event:

break

stream.close()

return True

def __del__(self):

pass

if __name__ == '__main__':

args = command_line_args(sys.argv[1:])

conn_setting = {'host': args.host, 'port': args.port, 'user': args.user, 'passwd': args.password, 'charset': 'utf8'}

binlog2sql = Binlog2sql(connection_settings=conn_setting, start_file=args.start_file, start_pos=args.start_pos,

end_file=args.end_file, end_pos=args.end_pos, start_time=args.start_time,

stop_time=args.stop_time, only_schemas=args.databases, only_tables=args.tables,

no_pk=args.no_pk, flashback=args.flashback, stop_never=args.stop_never,

back_interval=args.back_interval, only_dml=args.only_dml, sql_type=args.sql_type)

binlog2sql.process_binlog(ip=args.host,port=args.port)

数据库配置python脚本：db_help.py

# -*- coding:utf-8 -*-

# coding=utf-8

import datetime

import time

import types

import pymysql

class MYSQL:

def __init__(self,ip,port,database_name,type,timeout):

if database_name == "结果集DB名称":

self.host = ip

self.port = port

self.user = "db_user"

self.pwd = "db_password"

self.db = database_name

self.type = type

self.timeout=timeout

def GetConnect(self):

self.conn = pymysql.connect(host=self.host, port=self.port, user=self.user, passwd=self.pwd, db=self.db,

charset="utf8",read_timeout=self.timeout)

if self.type ==1:

cur = self.conn.cursor(pymysql.cursors.DictCursor)

else:

cur = self.conn.cursor()

return cur

def ExecQuery(self, sql,cur):

cur.execute(sql)

resList = cur.fetchall()

return resList

def ExecNonQuery(self, sql,cur):

cur.connection.autocommit(True)

cur.execute(sql)

def ms_close(self, cur):

self.conn.close()

cur.close()

运行脚本，进行数据收集：

python dbms_monitor_table_dml_stat.py -h 生产MySQL地址 -u DB用户 -p 密码 --stop-never --start-file 起始同步的binlog文件名 --start-pos 起始binlog的pos

六、补充

达到收集DML目的，也可以利用MySQL sys库schema_table_statistics。时间粒度可控性不强。

我之所以使用binlog2sql，收集DML只是顺手干的，要实现的目标扩展功能是准实时将表、主键、值、SQL和回滚SQL记录下来。这样方便研发排查记录的变更轨迹、为回滚需求提供自定义界面和准确地评估依据。