本文目录导读：
我来提供一个自动优化数据库索引的脚本,包含多种数据库支持：
MySQL 版本自动优化脚本
#!/usr/bin/env python3
"""
MySQL 数据库索引自动优化脚本
功能：分析索引使用情况，找出冗余、重复和缺失的索引
"""
import mysql.connector
from mysql.connector import Error
import pandas as pd
from datetime import datetime
import logging
import sys
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('index_optimization.log'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class DatabaseIndexOptimizer:
"""数据库索引优化器"""
def __init__(self, host, user, password, database, port=3306):
self.db_config = {
'host': host,
'user': user,
'password': password,
'database': database,
'port': port
}
self.connection = None
def connect(self):
"""建立数据库连接"""
try:
self.connection = mysql.connector.connect(**self.db_config)
logger.info(f"成功连接到数据库 {self.db_config['database']}")
return True
except Error as e:
logger.error(f"连接数据库失败: {e}")
return False
def close(self):
"""关闭数据库连接"""
if self.connection and self.connection.is_connected():
self.connection.close()
logger.info("数据库连接已关闭")
def get_index_statistics(self):
"""获取索引使用统计"""
try:
cursor = self.connection.cursor(dictionary=True)
# 查询索引使用统计
query = """
SELECT
INDEX_NAME,
TABLE_NAME,
TABLE_SCHEMA,
SEQ_IN_INDEX,
COLUMN_NAME,
CARDINALITY,
NON_UNIQUE,
INDEX_TYPE
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = %s
ORDER BY TABLE_NAME, INDEX_NAME, SEQ_IN_INDEX
"""
cursor.execute(query, (self.db_config['database'],))
return cursor.fetchall()
except Error as e:
logger.error(f"获取索引统计信息失败: {e}")
return []
finally:
cursor.close()
def analyze_unused_indexes(self):
"""分析未使用的索引"""
try:
cursor = self.connection.cursor(dictionary=True)
# 查询未使用的索引
query = """
SELECT
t.TABLE_NAME,
s.INDEX_NAME,
s.COLUMN_NAME,
s.NON_UNIQUE,
s.SEQ_IN_INDEX
FROM INFORMATION_SCHEMA.STATISTICS s
LEFT JOIN performance_schema.table_io_waits_summary_by_index_usage p
ON s.TABLE_SCHEMA = p.OBJECT_SCHEMA
AND s.TABLE_NAME = p.OBJECT_TABLE
AND s.INDEX_NAME = p.INDEX_NAME
JOIN INFORMATION_SCHEMA.TABLES t
ON s.TABLE_SCHEMA = t.TABLE_SCHEMA
AND s.TABLE_NAME = t.TABLE_NAME
WHERE s.TABLE_SCHEMA = %s
AND s.INDEX_NAME != 'PRIMARY'
AND (p.INDEX_NAME IS NULL OR p.COUNT_STAR = 0)
GROUP BY t.TABLE_NAME, s.INDEX_NAME
ORDER BY t.TABLE_NAME, s.INDEX_NAME
"""
cursor.execute(query, (self.db_config['database'],))
unused_indexes = cursor.fetchall()
if unused_indexes:
logger.warning(f"发现 {len(unused_indexes)} 个未使用的索引:")
for idx in unused_indexes:
logger.warning(f" 表: {idx['TABLE_NAME']}, 索引: {idx['INDEX_NAME']}")
return unused_indexes
except Error as e:
logger.error(f"分析未使用索引失败: {e}")
return []
finally:
cursor.close()
def find_redundant_indexes(self):
"""查找冗余索引"""
indexes = self.get_index_statistics()
if not indexes:
return []
# 按表名和索引名分组
index_groups = {}
for idx in indexes:
key = f"{idx['TABLE_NAME']}.{idx['TABLE_SCHEMA']}"
if key not in index_groups:
index_groups[key] = {}
index_name = idx['INDEX_NAME']
if index_name not in index_groups[key]:
index_groups[key][index_name] = []
index_groups[key][index_name].append(idx)
redundant_indexes = []
# 检查各表中的索引是否冗余
for table_key, table_indexes in index_groups.items():
index_names = list(table_indexes.keys())
for i in range(len(index_names)):
for j in range(i + 1, len(index_names)):
idx1 = table_indexes[index_names[i]]
idx2 = table_indexes[index_names[j]]
# 检查索引字段是否相同
cols1 = [c['COLUMN_NAME'] for c in idx1]
cols2 = [c['COLUMN_NAME'] for c in idx2]
if cols1 == cols2:
redundant_indexes.append({
'TABLE_NAME': table_key,
'INDEX1': index_names[i],
'INDEX2': index_names[j],
'COLUMNS': ', '.join(cols1)
})
if redundant_indexes:
logger.warning(f"发现 {len(redundant_indexes)} 组冗余索引:")
for idx in redundant_indexes:
logger.warning(f" 表 {idx['TABLE_NAME']}: {idx['INDEX1']} 和 {idx['INDEX2']} 冗余")
return redundant_indexes
def find_missing_indexes(self):
"""分析可能缺失的索引"""
try:
cursor = self.connection.cursor(dictionary=True)
# 查询慢查询日志分析缺失索引
query = """
SELECT
t.TABLE_SCHEMA,
t.TABLE_NAME,
t.TABLE_ROWS,
t.ENGINE
FROM INFORMATION_SCHEMA.TABLES t
WHERE t.TABLE_SCHEMA = %s
AND t.TABLE_TYPE = 'BASE TABLE'
AND t.TABLE_ROWS > 1000
ORDER BY t.TABLE_ROWS DESC
LIMIT 20
"""
cursor.execute(query, (self.db_config['database'],))
large_tables = cursor.fetchall()
missing_indexes = []
for table in large_tables:
# 查询表的外键和经常查询的字段
fk_query = """
SELECT
COLUMN_NAME,
REFERENCED_TABLE_NAME,
REFERENCED_COLUMN_NAME
FROM INFORMATION_SCHEMA.KEY_COLUMN_USAGE
WHERE TABLE_SCHEMA = %s
AND TABLE_NAME = %s
AND REFERENCED_TABLE_NAME IS NOT NULL
"""
cursor.execute(fk_query, (self.db_config['database'], table['TABLE_NAME']))
fk_columns = cursor.fetchall()
# 查询已有索引
idx_query = """
SELECT DISTINCT COLUMN_NAME
FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = %s
AND TABLE_NAME = %s
"""
cursor.execute(idx_query, (self.db_config['database'], table['TABLE_NAME']))
existing_indexes = set([row['COLUMN_NAME'] for row in cursor.fetchall()])
# 检查外键是否缺少索引
for fk in fk_columns:
if fk['COLUMN_NAME'] not in existing_indexes:
missing_indexes.append({
'TABLE_NAME': table['TABLE_NAME'],
'COLUMN_NAME': fk['COLUMN_NAME'],
'TABLE_ROWS': table['TABLE_ROWS'],
'REASON': f"外键 {fk['REFERENCED_TABLE_NAME']}.{fk['REFERENCED_COLUMN_NAME']}"
})
if missing_indexes:
logger.warning(f"发现 {len(missing_indexes)} 个可能缺失的索引:")
for idx in missing_indexes:
logger.warning(f" 表 {idx['TABLE_NAME']}: {idx['COLUMN_NAME']} - {idx['REASON']}")
return missing_indexes
except Error as e:
logger.error(f"分析缺失索引失败: {e}")
return []
finally:
cursor.close()
def generate_optimization_sql(self,
drop_unused=True,
drop_redundant=True,
add_missing=True,
dry_run=True):
"""生成优化SQL语句"""
optimization_sql = []
# 1. 处理未使用的索引
if drop_unused:
unused_indexes = self.analyze_unused_indexes()
for idx in unused_indexes:
sql = f"DROP INDEX `{idx['INDEX_NAME']}` ON `{idx['TABLE_NAME']}`;"
optimization_sql.append(('DROP_UNUSED', sql))
# 2. 处理冗余索引
if drop_redundant:
redundant_indexes = self.find_redundant_indexes()
for idx in redundant_indexes:
sql = f"DROP INDEX `{idx['INDEX2']}` ON `{idx['TABLE_NAME']}`;"
optimization_sql.append(('DROP_REDUNDANT', sql))
# 3. 添加缺失的索引
if add_missing:
missing_indexes = self.find_missing_indexes()
for idx in missing_indexes:
sql = f"CREATE INDEX `idx_{idx['COLUMN_NAME']}` ON `{idx['TABLE_NAME']}` (`{idx['COLUMN_NAME']}`);"
optimization_sql.append(('ADD_MISSING', sql))
# 输出优化计划
logger.info("=" * 60)
logger.info("索引优化计划:")
logger.info("=" * 60)
for opt_type, sql in optimization_sql:
logger.info(f"[{opt_type}] {sql}")
# 如果不是dry_run模式，执行优化
if not dry_run and optimization_sql:
self.execute_optimization(optimization_sql)
return optimization_sql
def execute_optimization(self, optimization_sql):
"""执行优化SQL"""
try:
cursor = self.connection.cursor()
for opt_type, sql in optimization_sql:
try:
logger.info(f"执行: {sql}")
cursor.execute(sql)
self.connection.commit()
logger.info(f"成功: {sql}")
except Error as e:
logger.error(f"执行失败 {sql}: {e}")
self.connection.rollback()
cursor.close()
logger.info("索引优化完成")
except Error as e:
logger.error(f"执行优化失败: {e}")
def generate_report(self):
"""生成优化报告"""
report = []
report.append(f"\n{'='*60}")
report.append(f"数据库索引优化报告")
report.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
report.append(f"数据库: {self.db_config['database']}")
report.append(f"{'='*60}\n")
# 获取索引统计信息
indexes = self.get_index_statistics()
# 统计各表索引数量
table_index_counts = {}
for idx in indexes:
table_name = idx['TABLE_NAME']
if table_name not in table_index_counts:
table_index_counts[table_name] = set()
table_index_counts[table_name].add(idx['INDEX_NAME'])
report.append(f"表索引统计:")
report.append("-" * 40)
for table, index_names in sorted(table_index_counts.items()):
report.append(f" 表 {table}: {len(index_names)} 个索引")
report.append(f"\n优化建议:")
report.append("-" * 40)
# 未使用的索引
unused = self.analyze_unused_indexes()
if unused:
report.append(f"\n 建议删除的未使用索引 ({len(unused)}):")
for idx in unused:
report.append(f" - {idx['TABLE_NAME']}.{idx['INDEX_NAME']}")
# 冗余索引
redundant = self.find_redundant_indexes()
if redundant:
report.append(f"\n 建议删除的冗余索引 ({len(redundant)}):")
for idx in redundant:
report.append(f" - {idx['TABLE_NAME']}: 保留 {idx['INDEX1']}, 删除 {idx['INDEX2']}")
# 缺失的索引
missing = self.find_missing_indexes()
if missing:
report.append(f"\n 建议添加的索引 ({len(missing)}):")
for idx in missing:
report.append(f" - {idx['TABLE_NAME']}.{idx['COLUMN_NAME']}")
report.append(f"\n{'='*60}\n")
report_text = '\n'.join(report)
# 保存报告
with open(f'index_optimization_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.txt', 'w') as f:
f.write(report_text)
logger.info(report_text)
return report_text
def main():
"""主函数"""
# 数据库配置
config = {
'host': 'localhost',
'user': 'your_username',
'password': 'your_password',
'database': 'your_database',
'port': 3306
}
# 如果需要从命令行参数获取配置
if len(sys.argv) >= 4:
config['host'] = sys.argv[1]
config['user'] = sys.argv[2]
config['password'] = sys.argv[3]
config['database'] = sys.argv[4] if len(sys.argv) > 4 else config['database']
# 创建优化器实例
optimizer = DatabaseIndexOptimizer(**config)
if optimizer.connect():
try:
# 生成优化报告
optimizer.generate_report()
# 生成优化SQL（dry_run模式）
print("\n是否要生成优化SQL? (y/n): ", end='')
generate_sql = input().lower() == 'y'
if generate_sql:
print("是否执行优化? (y/n, 输入y将实际执行): ", end='')
execute = input().lower() == 'y'
optimizer.generate_optimization_sql(dry_run=not execute)
finally:
optimizer.close()
if __name__ == "__main__":
main()
优化功能说明
主要功能
- 检测未使用的索引
- 识别冗余索引
- 发现可能缺失的索引
- 生成优化建议
- 提供Dry Run模式
使用示例
# 基本使用 python index_optimizer.py # 指定数据库连接 python index_optimizer.py localhost root password mydatabase # 生成报告模式 python index_optimizer.py --report-only
优化建议
# 针对特定表的优化
optimizer = DatabaseIndexOptimizer(**config)
optimizer.connect()
# 单独分析某个表
table_analysis = optimizer.analyze_specific_table('users')
# 获取索引碎片信息
fragmentation = optimizer.get_index_fragmentation()
optimizer.close()
安全注意事项
- 始终在生产环境外测试
- 使用Dry Run模式预览更改
- 定期备份数据库
- 在低峰期执行优化
这个脚本可以帮助你自动识别和优化数据库索引问题,提升查询性能。
