自动优化数据库索引的脚本

本文目录导读：

1. MySQL 版本自动优化脚本
2. 优化功能说明

我来提供一个自动优化数据库索引的脚本,包含多种数据库支持：

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模式预览更改
• 定期备份数据库
• 在低峰期执行优化

这个脚本可以帮助你自动识别和优化数据库索引问题,提升查询性能。