针对 binlog MIXED 格式对表的增删改统计分析

mysqlbinlog --base64-output=DECODE-ROWS -v mysql-bin.000001 > binlog_000001.sql

#!/usr/bin/env python3
import re
import sys
from collections import defaultdict
 
def analyze_binlog(file_path):
    """
    分析由 mysqlbinlog -v 生成的文本文件，统计每个表的增删改操作次数。
    """
    # 使用字典保存统计结果，格式： {‘database.table‘： {'INSERT': 0, 'UPDATE': 0, 'DELETE': 0}}
    statistics = defaultdict(lambda: defaultdict(int))
 
    # 状态变量
    current_db = None
    current_table = None
    current_event = None  # ‘INSERT‘, ‘UPDATE‘, ‘DELETE‘
    in_rows_event = False
 
    # 正则表达式模式
    # 匹配 STATEMENT 格式的 SQL 语句开头，例如： ‘UPDATE `db`.`tbl`‘ 或 ‘INSERT INTO `db`.`tbl`‘
    stmt_pattern = re.compile(r'^(UPDATE|INSERT INTO|DELETE FROM)\s+`?(\w+)`?\.`?(\w+)`?', re.IGNORECASE)
    # 匹配 ROW 格式的 Table_map 事件，例如： ‘### Table_map: `db`.`tbl` mapped to number 240‘
    table_map_pattern = re.compile(r'^### Table_map: `?(\w+)`?\.`?(\w+)`? mapped to number')
    # 匹配 ROW 格式的伪 SQL 行，例如： ‘### INSERT INTO `db`.`tbl`‘
    row_event_start_pattern = re.compile(r'^### (INSERT INTO|UPDATE|DELETE FROM)')
    # 匹配 ROW 格式中标记操作类型的行，例如： ‘### UPDATE `db`.`tbl`‘ 是更精确的表名来源
    row_event_table_pattern = re.compile(r'^### (INSERT INTO|UPDATE|DELETE FROM) `?(\w+)`?\.`?(\w+)`?')
 
    with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
        for line in f:
            line = line.strip()
 
            # 1. 尝试匹配 STATEMENT 格式的 SQL
            stmt_match = stmt_pattern.match(line)
            if stmt_match:
                # 找到一条 STATEMENT 格式的 DML
                event_type = stmt_match.group(1).upper().split()[0]  # 取第一个单词
                db_name = stmt_match.group(2)
                table_name = stmt_match.group(3)
                key = f"{db_name}.{table_name}"
                statistics[key][event_type] += 1
                # 重置 ROW 事件状态
                in_rows_event = False
                current_event = None
                continue
 
            # 2. 尝试匹配 ROW 格式的 Table_map 事件（它先于行数据出现）
            table_map_match = table_map_pattern.match(line)
            if table_map_match:
                current_db = table_map_match.group(1)
                current_table = table_map_match.group(2)
                continue
 
            # 3. 尝试匹配 ROW 格式事件的开始（伪 SQL 开头）
            row_start_match = row_event_start_pattern.match(line)
            if row_start_match:
                in_rows_event = True
                event_type = row_start_match.group(1).upper().split()[0]
                current_event = event_type
                # 尝试从这行获取更精确的表名（有时 Table_map 可能不是最近的一个）
                row_table_match = row_event_table_pattern.match(line)
                if row_table_match:
                    current_db = row_table_match.group(2)
                    current_table = row_table_match.group(3)
                continue
 
            # 4. 如果在 ROW 事件内部，并且遇到 ‘###‘ 开头的行，我们认为是数据行，开始计数
            if in_rows_event and line.startswith('### '):
                # 忽略 SET WHERE 等细节行，只统计有表名的开始行触发的计数
                # 但为了避免重复计数，我们更精确地判断：通常一条 ROW 事件只有一个操作。
                # 所以计数应该在遇到 ‘# at‘（下一个事件位置）时进行。这里我们简化处理。
                # 更准确的做法是：在遇到下一个 ‘# at‘ 时，为 current_db/current_table/current_event 计数一次。
                pass
 
            # 5. 匹配事件结束标记（下一个事件的开始）
            if line.startswith('# at ') and in_rows_event and current_db and current_table and current_event:
                key = f"{current_db}.{current_table}"
                statistics[key][current_event] += 1
                # 重置状态
                in_rows_event = False
                current_event = None
                # 注意：不清除 current_db/current_table，因为下一个事件可能还是同一个表
 
    return statistics
 
if __name__ == '__main__':
    if len(sys.argv) != 2:
        print("Usage: python analyze_binlog.py <binlog_text_file>")
        sys.exit(1)
 
    file_path = sys.argv[1]
    stats = analyze_binlog(file_path)
 
    # 打印结果
    print("Binlog 操作统计报告:")
    print("-" * 50)
    for table, counts in sorted(stats.items()):
        total = sum(counts.values())
        print(f"表: {table} (总操作数: {total})")
        for op_type, count in sorted(counts.items()):
            if count > 0:
                print(f"  {op_type}: {count}")
        print()

python3 analyze_binlog.py binlog_000001.sql