MySQL慢查询日志分析脚本

#!/usr/bin/env python3 # -*- coding: utf-8 -*- ''' MySQL慢查询日志分析脚本 用途：分析慢查询日志，识别性能瓶颈，给出优化建议 使用方法：python3 mysql_slow_log_analyzer.py /path/to/slow.log ''' import sys import re from collections import defaultdict, Counter from datetime import datetime def parse_slow_log(filepath): '''解析慢查询日志''' queries = [] current_query = None query_time = 0 lock_time = 0 rows_examined = 0 rows_sent = 0 timestamp = None with open(filepath, 'r', encoding='utf-8', errors='ignore') as f: for line in f: line = line.strip() # 新查询开始 if line.startswith('# Time:') or line.startswith('# Query_time:'): if current_query: queries.append({ 'query': current_query.strip(), 'query_time': query_time, 'lock_time': lock_time, 'rows_examined': rows_examined, 'rows_sent': rows_sent, 'timestamp': timestamp }) current_query = '' query_time = 0 lock_time = 0 rows_examined = 0 rows_sent = 0 if line.startswith('# Time:'): time_str = line.replace('# Time:', '').strip() try: timestamp = datetime.strptime(time_str, '%Y-%m-%dT%H:%M:%S.%fZ') except: timestamp = None elif line.startswith('# Query_time:'): match = re.search(r'Query_time:\s+([\d.]+)\s+Lock_time:\s+([\d.]+)', line) if match: query_time = float(match.group(1)) lock_time = float(match.group(2)) match2 = re.search(r'Rows_examined:\s+(\d+)\s+Rows_sent:\s+(\d+)', line) if match2: rows_examined = int(match2.group(1)) rows_sent = int(match2.group(2)) elif line and not line.startswith('#') and not line.startswith('use ') and not line.startswith('SET '): current_query += line + ' ' # 最后一个查询 if current_query: queries.append({ 'query': current_query.strip(), 'query_time': query_time, 'lock_time': lock_time, 'rows_examined': rows_examined, 'rows_sent': rows_sent, 'timestamp': timestamp }) return queries def normalize_query(query): '''标准化查询，用于分组统计''' # 移除注释 query = re.sub(r'/\*.*?\*/', '', query) # 替换数字为? query = re.sub(r'\d+', '?', query) # 替换字符串为? query = re.sub(r"'[^']*'", "'?'", query) query = re.sub(r'"[^"]*"', '"?"', query) # 移除多余空格 query = re.sub(r'\s+', ' ', query).strip() return query[:200] def analyze_queries(queries): '''分析查询统计''' if not queries: print("未找到慢查询记录") return print("=" * 60) print("MySQL慢查询日志分析报告") print("=" * 60) print(f"分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"慢查询总数: {len(queries)}") print() # 总体统计 total_time = sum(q['query_time'] for q in queries) avg_time = total_time / len(queries) max_time = max(q['query_time'] for q in queries) total_rows_examined = sum(q['rows_examined'] for q in queries) print("【1. 总体统计】") print(f" 总执行时间: {total_time:.2f}秒") print(f" 平均执行时间: {avg_time:.2f}秒") print(f" 最大执行时间: {max_time:.2f}秒") print(f" 总扫描行数: {total_rows_examined:,}") print() # 按查询类型分组 print("【2. 查询类型分布】") query_types = Counter() for q in queries: query_upper = q['query'].upper() if query_upper.startswith('SELECT'): query_types['SELECT'] += 1 elif query_upper.startswith('INSERT'): query_types['INSERT'] += 1 elif query_upper.startswith('UPDATE'): query_types['UPDATE'] += 1 elif query_upper.startswith('DELETE'): query_types['DELETE'] += 1 else: query_types['OTHER'] += 1 for qtype, count in query_types.most_common(): pct = count / len(queries) * 100 print(f" {qtype}: {count}次 ({pct:.1f}%)") print() # Top 10 最慢查询 print("【3. Top 10 最慢查询】") sorted_queries = sorted(queries, key=lambda x: x['query_time'], reverse=True) for i, q in enumerate(sorted_queries[:10], 1): print(f" {i}. 执行时间: {q['query_time']:.2f}秒, 扫描行数: {q['rows_examined']:,}") print(f" SQL: {q['query'][:100]}...") print() # 高频慢查询（标准化后分组） print("【4. 高频慢查询 Top 10】") query_groups = defaultdict(list) for q in queries: normalized = normalize_query(q['query']) query_groups[normalized].append(q) sorted_groups = sorted(query_groups.items(), key=lambda x: len(x[1]), reverse=True) for i, (normalized, group) in enumerate(sorted_groups[:10], 1): avg_group_time = sum(q['query_time'] for q in group) / len(group) print(f" {i}. 出现次数: {len(group)}次, 平均时间: {avg_group_time:.2f}秒") print(f" SQL模式: {normalized[:100]}...") print() # 优化建议 print("【5. 优化建议】") print(" 1. 索引优化:") print(" - 检查WHERE条件字段是否有索引") print(" - 避免在索引列上使用函数或运算") print(" - 考虑使用覆盖索引，减少回表") print(" 2. 查询优化:") print(" - 避免SELECT *，只查询需要的字段") print(" - 使用LIMIT限制返回行数") print(" - 大表分页使用游标分页而非OFFSET") print(" 3. 表结构优化:") print(" - 大表考虑分区") print(" - 定期分析表: ANALYZE TABLE") print(" - 检查是否有冗余索引") print(" 4. 配置优化:") print(" - 适当调整innodb_buffer_pool_size") print(" - 检查slow_query_log阈值是否合理") print() print("=" * 60) print("分析完成") print("=" * 60) if __name__ == '__main__': if len(sys.argv) < 2: print("使用方法: python3 mysql_slow_log_analyzer.py /path/to/slow.log") sys.exit(1) log_file = sys.argv[1] try: queries = parse_slow_log(log_file) analyze_queries(queries) except FileNotFoundError: print(f"错误: 文件不存在 - {log_file}") sys.exit(1) except Exception as e: print(f"分析出错: {e}") sys.exit(1)

[{"name": "LOG_FILE", "label": "\u6162\u67e5\u8be2\u65e5\u5fd7\u8def\u5f84", "default": "/var/log/mysql/slow.log"}]