数据库慢查询的工业化治理——从慢日志分析到执行计划自动优化的全链路方案一、慢查询不是改完索引就结束——它是需要持续运营的系统性工程在一次大促前的数据库巡检中DBA 发现orders表上有 12 条慢查询其中 8 条可以通过加索引解决。加完索引后慢查询从 45 条/小时降到了 6 条/小时——看似解决了问题。一个月后慢查询又回到了 38 条/小时。原因业务上线了新功能新增ORDER BY refund_time没人告诉 DBA。之前的索引中有 2 个已经不再被任何查询使用冗余索引占着空间还拖慢写入。users表从 500 万行增长到 2000 万行原来能走索引的范围扫描因为数据量变大变慢了。慢查询治理不是一次性的加索引动作而是需要持续运营的系统性工程——包括慢日志自动采集、执行计划分析、索引推荐和影响评估。二、慢查询治理的全链路流程flowchart LR A[慢查询日志br/MySQL Slow Log] -- B[日志采集br/Filebeat/Fluentd] B -- C[解析与聚合br/pt-query-digest] C -- D{分类分析} D -- E[新增慢查询br/执行计画缺失索引] D -- F[已有慢查询恶化br/数据增长导致] D -- G[冗余索引检测br/从未使用的索引] E -- H[索引推荐引擎br/基于 EXPLAIN 分析] F -- I[执行计划对比br/历史 vs 当前] G -- J[索引下线建议br/避免写入负担] H -- K[影响评估br/索引大小/写入影响] I -- K K -- L{自动执行或人工审核} L --|低风险| M[自动创建索引br/灰度放量] L --|高风险| N[人工 Reviewbr/DBA 审批]2.1 慢日志的三分类治理的第一步是不把所有慢查询混为一谈。按根因分成三类类别特征治理手段无合适索引EXPLAIN中typeALL全表扫描索引推荐引擎分析 创建索引索引失效EXPLAIN中有Using index condition但rows极大重建索引统计信息 / 重写 SQL数据量增长去年同期同样 SQL 的执行时间增长 50%分表/归档/读写分离锁等待Lock_time占比 20%减小事务范围、拆分大事务三、自动化治理管道的生产级实现3.1 慢日志解析与聚合# slow_query_analyzer.py —— 慢查询日志分析管道 import re import hashlib from collections import defaultdict from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional, Dict import subprocess import json dataclass class SlowQuery: 单条慢查询的解析结果 # 慢日志中的原始时间 query_time: datetime # 锁定时间秒 lock_time: float # 扫描行数 rows_examined: int # 返回行数 rows_sent: int # 查询指纹参数化后的模板 fingerprint: str # 原始 SQL raw_sql: str # 数据库 database: str # 用户 user: str dataclass class AggregatedQuery: 聚合后的慢查询报告 fingerprint: str # 示例 SQL sample_sql: str # 出现次数 count: int # 总耗时秒 total_time: float # 平均扫描行数 avg_rows_examined: float # 扫描/返回比越高越浪费 scan_return_ratio: float # 是否新增之前未出现过 is_new: bool False class SlowQueryAnalyzer: 慢查询分析器 使用 pt-query-digest 作为核心解析引擎 自动参数化 SQL 并聚合相同模式的慢查询。 def __init__(self, slow_log_path: str): self.slow_log_path slow_log_path def parse( self, since: Optional[str] None ) - List[AggregatedQuery]: 解析慢日志返回聚合后的 Top-N 慢查询 Args: since: 起始时间格式 2024-07-01 00:00:00 None 表示全部 # 1. 使用 pt-query-digest 解析慢日志 # --no-report: 不生成 HTML 报告更快 # --output json: 输出 JSON 格式 cmd [ pt-query-digest, --no-report, --output, json, --limit, 50, # Top-50 慢查询 --order-by, Query_time:sum, # 按总耗时排序 ] if since: cmd.extend([--since, since]) cmd.append(self.slow_log_path) result subprocess.run( cmd, capture_outputTrue, textTrue, timeout60, # 大慢日志文件可能需要时间 ) # 2. 解析 JSON 输出 data json.loads(result.stdout) # 3. 转换为聚合对象 queries [] for item in data.get(classes, []): metrics item.get(metrics, {}) rows_examined metrics.get( Rows_examine, {}).get(avg, 0) rows_sent metrics.get( Rows_sent, {}).get(avg, 0) ratio ( rows_examined / rows_sent if rows_sent 0 else float(inf) ) queries.append(AggregatedQuery( fingerprintitem.get(fingerprint, ), sample_sqlitem.get(example, {}).get( query, ), countmetrics.get( Query_count, {}).get(sum, 0), total_timemetrics.get( Query_time, {}).get(sum, 0), avg_rows_examinedrows_examined, scan_return_ratioratio, )) return queries3.2 索引推荐引擎# index_advisor.py —— 索引推荐引擎 import re from typing import List, Dict, Optional, Tuple from dataclasses import dataclass dataclass class IndexRecommendation: 索引推荐结果 table: str columns: List[str] # 推荐的索引列按顺序 sql_fingerprint: str # 对应的慢查询指纹 estimated_improvement: float # 预估提升倍数 # 风险等级 risk: str # low | medium | high # 索引大小预估MB estimated_size_mb: float # 对写入性能的影响预估 write_impact: str # low | medium | high class IndexAdvisor: 索引推荐引擎 核心流程 1. 对每条慢查询执行 EXPLAIN获取当前执行计划 2. 分析 WHERE/JOIN/ORDER BY 列的使用模式 3. 生成候选索引 4. 通过 EXPLAIN 验证候选索引的效果 def __init__(self, db_conn): self.db_conn db_conn def analyze( self, query: AggregatedQuery ) - List[IndexRecommendation]: 分析一条聚合慢查询生成索引推荐 Args: query: 聚合后的慢查询 Returns: 索引推荐列表可能有多个候选索引 recommendations [] # 1. 执行 EXPLAIN 看当前执行计划 explain_result self._explain(query.sample_sql) if explain_result is None: return [] # 2. 如果已经是好计划跳过 # type 为 ALL/index 表示全表/全索引扫描 if explain_result[type] not in ( ALL, index, index_merge): return [] # 扫描类型已经不错 # 3. 提取 SQL 中的过滤和排序条件 conditions self._extract_conditions( query.sample_sql) # 4. 生成候选索引 # 规则 # - WHERE 等值条件 → 索引前缀 # - WHERE 范围条件 → 索引中间或末尾 # - ORDER BY → 索引最后列避免 filesort # - GROUP BY → 可以用于索引前缀 candidate self._build_candidate_index( conditions) if not candidate: return [] table explain_result.get(table, unknown) # 5. 模拟索引效果 # 通过 USE INDEX 或 FORCE INDEX 的 EXPLAIN # 但 MySQL 不支持不存在的索引 EXPLAIN # 故使用启发式方法预估 improvement self._estimate_improvement( query, explain_result) # 6. 计算索引大小 estimated_size self._estimate_index_size( table, candidate) recommendations.append(IndexRecommendation( tabletable, columnscandidate, sql_fingerprintquery.fingerprint, estimated_improvementimprovement, riskself._assess_risk(table, candidate), estimated_size_mbestimated_size, write_impactself._assess_write_impact( table, candidate), )) return recommendations def _explain(self, sql: str) - Optional[Dict]: 执行 EXPLAIN 并解析结果 try: cursor self.db_conn.cursor() cursor.execute(fEXPLAIN {sql}) columns [col[0] for col in cursor.description] row cursor.fetchone() if row: return dict(zip(columns, row)) except Exception: pass return None def _extract_conditions( self, sql: str ) - Dict[str, List[str]]: 从 SQL 中提取过滤/排序条件 简化实现使用正则提取 WHERE 和 ORDER BY 子句中的列名 实际生产环境应使用 SQL 解析器如 sqlparse conditions { where_eq: [], # WHERE col ? where_range: [], # WHERE col ? / BETWEEN order_by: [], # ORDER BY col group_by: [], # GROUP BY col join_on: [], # JOIN ... ON col ? } # 提取 WHERE 子句 where_match re.search( rWHERE\s(.?)(?:ORDER|GROUP|LIMIT|$), sql, re.IGNORECASE | re.DOTALL) if where_match: where_clause where_match.group(1) # 等值条件: col ? eq_cols re.findall( r(\w\.?\w*)\s*\s*[?\], where_clause) conditions[where_eq] eq_cols # 范围条件: col / / BETWEEN range_cols re.findall( r(\w\.?\w*)\s*(?|?|BETWEEN)\s*, where_clause) conditions[where_range] [ c[0] for c in range_cols] # 提取 ORDER BY order_match re.search( rORDER\sBY\s(.?)(?:LIMIT|$), sql, re.IGNORECASE) if order_match: order_cols re.findall( r(\w\.?\w*), order_match.group(1)) conditions[order_by] order_cols return conditions def _build_candidate_index( self, conditions: Dict[str, List[str]] ) - List[str]: 根据条件构建候选索引列顺序 索引列顺序原则被称作Star Schema Rule 1. WHERE 等值条件列选择性最高的在前 2. WHERE 范围条件列最多 1 个之后的列不再用于过滤 3. ORDER BY / GROUP BY 列避免 filesort/temporary candidate [] # 1. 等值条件 → 索引最前面 candidate.extend(conditions.get(where_eq, [])) # 2. 范围条件 → 只取 1 个最多 range_cols conditions.get(where_range, []) if range_cols: candidate.append(range_cols[0]) # 3. ORDER BY 列 → 避免 filesort order_cols conditions.get(order_by, []) for col in order_cols: if col not in candidate: candidate.append(col) # 去重保留第一次出现的位置 seen set() unique [] for col in candidate: if col not in seen: seen.add(col) unique.append(col) return unique[:5] # 最多 5 列避免索引过大 def _estimate_improvement( self, query: AggregatedQuery, explain: Dict, ) - float: 预估加索引后的性能提升 简化模型 - 如果当前是全表扫描typeALL且扫描/返回比 100 → 预估提升 扫描/返回比保守取 10 倍上限 - 如果是索引扫描typeindex → 预估提升 rows/5保守估计 ratio query.scan_return_ratio if explain.get(type) ALL: return min(ratio, 100) return min(ratio / 5, 20) def _assess_risk( self, table: str, columns: List[str] ) - str: 评估创建索引的风险 # 规则示例 # - 大表 1000 万行新索引 → 高风险锁表时间长 # - 已有多个索引的同列组合 → 低风险 # - 生产高峰时段 → 中风险 cursor self.db_conn.cursor() cursor.execute( fSELECT TABLE_ROWS FROM INFORMATION_SCHEMA.TABLES fWHERE TABLE_NAME {table}) row cursor.fetchone() if row and row[0] and row[0] 10_000_000: return high return low def _estimate_index_size( self, table: str, columns: List[str] ) - float: 预估索引大小MB cursor self.db_conn.cursor() cursor.execute( fSELECT AVG_ROW_LENGTH, TABLE_ROWS fFROM INFORMATION_SCHEMA.TABLES fWHERE TABLE_NAME {table}) row cursor.fetchone() if not row or not row[0]: return 0 # 粗略公式索引大小 ≈ 列数 × 8 bytes × 行数 / 1024^2 return len(columns) * 8 * (row[1] or 0) / (1024 * 1024)3.3 冗余索引检测# unused_index_detector.py —— 冗余索引检测 def detect_unused_indexes(db_conn) - List[Dict]: 从 performance.schema 检测从未使用的索引 原理MySQL 5.7 的 performance_schema.table_io_waits_summary_by_index_usage 追踪了每个索引的 IO 统计。COUNT_STAR 0 表示从未使用。 sql SELECT OBJECT_SCHEMA AS db_name, OBJECT_NAME AS table_name, INDEX_NAME AS index_name, COUNT_STAR AS access_count, -- 主键和唯一约束的索引不能随意删除 IF(INDEX_NAME PRIMARY, NO, MAYBE) AS can_drop FROM performance_schema.table_io_waits_summary_by_index_usage WHERE OBJECT_SCHEMA NOT IN (mysql, performance_schema, sys) -- 从未使用 AND COUNT_STAR 0 -- 排除自增计数器 AND INDEX_NAME NOT LIKE auto_gen% ORDER BY OBJECT_SCHEMA, OBJECT_NAME cursor db_conn.cursor() cursor.execute(sql) columns [col[0] for col in cursor.description] return [dict(zip(columns, row)) for row in cursor.fetchall()]四、自动索引创建的工程风险4.1 索引不是免费的每增加一个索引写入性能下降约 5%15%取决于索引列数和行数。增加存储空间约为表大小的 10%50%取决于索引列。可能影响查询优化器的选择统计信息变化可能导致其他查询的执行计划变差。自动索引推荐必须附带写入影响评估对写入密集型表每秒 1000 INSERT/UPDATE任何索引都要经过人工审核。4.2 在线 DDL 的陷阱MySQL 5.6 支持 Online DDLALGORITHMINPLACE但在以下情况下仍会锁表添加全文索引FULLTEXT。添加空间索引SPATIAL。某些字符集和排序规则的变更。生产环境创建索引时始终指定ALTER TABLE orders ADD INDEX idx_xxx (cols) ALGORITHMINPLACE, LOCKNONE;并在执行前检查是否有未提交的长事务——长事务会阻塞 DDL 的元数据锁获取。4.3 索引推荐引擎的误判EXPLAIN 的预估行数rows与实际行数可能存在数量级差异——特别是当统计信息过时时innodb_stats_persistent未更新。索引推荐引擎给出的预估提升 50 倍可能是基于错误的统计信息。五、总结慢查询治理的工业化路径是从问题驱动的一次性修复到数据驱动的持续运营自动采集pt-query-digest 定期每日分析慢日志参数化 SQL 并聚合相同模式。三分类处理无索引 → 索引推荐引擎数据增长 → 分表/归档锁等待 → 事务优化。索引推荐 影响评估EXPLAIN 分析生成候选索引 预估写入影响 冗余索引检测。执行安全Online DDL 长事务检查 灰度放量先在从库创建验证再上主库。