如何编写优化数据库索引脚本

本文目录导读：

1. 索引检查脚本
2. 索引优化脚本
3. 自动化监控脚本
4. 索引维护脚本
5. 最佳实践建议

我来详细介绍如何编写优化数据库索引的脚本，涵盖MySQL、PostgreSQL等主流数据库。

索引检查脚本

MySQL索引分析脚本

-- 查找未使用的索引
SELECT 
    OBJECT_SCHEMA AS '数据库',
    OBJECT_NAME AS '表名',
    INDEX_NAME AS '索引名',
    COUNT_STAR AS '使用次数',
    LAST_USED AS '最后使用时间'
FROM performance_schema.table_index_usage_statistics
WHERE INDEX_NAME IS NOT NULL
  AND COUNT_STAR = 0
ORDER BY OBJECT_SCHEMA, OBJECT_NAME;
-- 查找重复索引
SELECT 
    t.TABLE_SCHEMA,
    t.TABLE_NAME,
    GROUP_CONCAT(DISTINCT i.INDEX_NAME SEPARATOR ', ') AS '重复索引'
FROM information_schema.TABLES t
JOIN information_schema.STATISTICS i ON t.TABLE_SCHEMA = i.TABLE_SCHEMA 
    AND t.TABLE_NAME = i.TABLE_NAME
WHERE t.TABLE_SCHEMA NOT IN ('mysql', 'performance_schema', 'information_schema')
GROUP BY t.TABLE_SCHEMA, t.TABLE_NAME, i.COLUMN_NAME
HAVING COUNT(DISTINCT i.INDEX_NAME) > 1;
-- 查找低效索引（索引基数低）
SELECT 
    TABLE_SCHEMA,
    TABLE_NAME,
    INDEX_NAME,
    COLUMN_NAME,
    CARDINALITY,
    (SELECT COUNT(*) FROM information_schema.COLUMNS c 
     WHERE c.TABLE_SCHEMA = s.TABLE_SCHEMA 
       AND c.TABLE_NAME = s.TABLE_NAME) AS '总行数'
FROM information_schema.STATISTICS s
WHERE TABLE_SCHEMA NOT IN ('mysql', 'performance_schema', 'information_schema')
  AND CARDINALITY < 10
ORDER BY CARDINALITY;

PostgreSQL索引分析脚本

-- 查找未使用的索引
SELECT
    schemaname || '.' || tablename AS "表名",
    indexrelname AS "索引名",
    idx_scan AS "扫描次数",
    idx_tup_read AS "读取元组数",
    idx_tup_fetch AS "获取元组数"
FROM pg_stat_user_indexes
WHERE idx_scan = 0
ORDER BY schemaname, tablename;
-- 查找重复索引
SELECT
    a.indrelid::regclass AS "表名",
    array_agg(a.indexrelid::regclass) AS "重复索引"
FROM pg_index a
JOIN pg_index b ON a.indrelid = b.indrelid
    AND a.indexrelid != b.indexrelid
    AND a.indkey = b.indkey
GROUP BY a.indrelid;
-- 查找碎片化索引
SELECT
    schemaname || '.' || tablename AS "表名",
    indexrelname AS "索引名",
    pg_size_pretty(pg_relation_size(indexrelid)) AS "索引大小",
    avg_leaf_density AS "叶密度",
    leaf_fragmentation AS "碎片率"
FROM pgstatindex('索引名');

索引优化脚本

MySQL索引优化

-- 创建索引优化建议存储过程
DELIMITER //
CREATE PROCEDURE optimize_indexes(IN db_name VARCHAR(64))
BEGIN
    DECLARE done INT DEFAULT FALSE;
    DECLARE tbl_name VARCHAR(64);
    DECLARE cur CURSOR FOR 
        SELECT TABLE_NAME 
        FROM information_schema.TABLES 
        WHERE TABLE_SCHEMA = db_name 
          AND TABLE_TYPE = 'BASE TABLE';
    DECLARE CONTINUE HANDLER FOR NOT FOUND SET done = TRUE;
    OPEN cur;
    read_loop: LOOP
        FETCH cur INTO tbl_name;
        IF done THEN
            LEAVE read_loop;
        END IF;
        -- 重建表索引
        SET @sql = CONCAT('ALTER TABLE ', db_name, '.', tbl_name, ' ENGINE=InnoDB');
        PREPARE stmt FROM @sql;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;
        -- 分析表更新统计信息
        SET @sql = CONCAT('ANALYZE TABLE ', db_name, '.', tbl_name);
        PREPARE stmt FROM @sql;
        EXECUTE stmt;
        DEALLOCATE PREPARE stmt;
    END LOOP;
    CLOSE cur;
END//
DELIMITER ;
-- 添加必要索引的脚本生成
SELECT CONCAT(
    'CREATE INDEX idx_', 
    TABLE_NAME, '_', 
    GROUP_CONCAT(COLUMN_NAME ORDER BY ORDINAL_POSITION SEPARATOR '_'),
    ' ON ', TABLE_SCHEMA, '.', TABLE_NAME, 
    ' (', GROUP_CONCAT(COLUMN_NAME ORDER BY ORDINAL_POSITION), ');'
) AS create_index_sql
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = 'your_database'
  AND COLUMN_KEY = ''  -- 没有索引的列
  AND (DATA_TYPE IN ('int', 'bigint', 'varchar', 'char')
       OR COLUMN_NAME LIKE '%id%'
       OR COLUMN_NAME LIKE '%fk%')
GROUP BY TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME;

PostgreSQL索引优化

-- 重建索引函数
CREATE OR REPLACE FUNCTION rebuild_indexes(schema_name TEXT)
RETURNS TABLE(index_name TEXT, status TEXT) AS $$
DECLARE
    idx RECORD;
BEGIN
    FOR idx IN 
        SELECT indexrelid::regclass::text AS idx_name
        FROM pg_stat_user_indexes
        WHERE schemaname = schema_name
          AND idx_scan > 0  -- 只重建使用的索引
    LOOP
        BEGIN
            EXECUTE format('REINDEX INDEX %I.%I', schema_name, idx.idx_name);
            RETURN QUERY SELECT idx.idx_name, 'success'::TEXT;
        EXCEPTION WHEN OTHERS THEN
            RETURN QUERY SELECT idx.idx_name, SQLERRM::TEXT;
        END;
    END LOOP;
END;
$$ LANGUAGE plpgsql;
-- 创建覆盖索引建议
SELECT 
    schemaname,
    tablename,
    'CREATE INDEX CONCURRENTLY idx_' || tablename || '_covering ON ' || 
    schemaname || '.' || tablename || ' (' || 
    string_agg(column_name, ', ' ORDER BY column_position) || 
    ') INCLUDE (' || 
    string_agg(include_column, ', ') || ');' AS create_index_sql
FROM (
    SELECT 
        schemaname,
        tablename,
        column_name,
        column_position,
        include_column
    FROM (
        SELECT 
            schemaname,
            tablename,
            column_name,
            ordinal_position AS column_position,
            NULL AS include_column,
            COUNT(*) OVER (PARTITION BY schemaname, tablename) AS col_count
        FROM pg_catalog.pg_indexes i
        JOIN information_schema.columns c 
            ON c.table_schema = i.schemaname 
            AND c.table_name = i.tablename
        WHERE i.indexdef LIKE '%WHERE%'  -- 只处理条件索引
    ) sub
) sub2
GROUP BY schemaname, tablename;

自动化监控脚本

Python自动化索引优化脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import mysql.connector
from datetime import datetime
import json
class IndexOptimizer:
    def __init__(self, host, user, password, database):
        self.connection = mysql.connector.connect(
            host=host,
            user=user,
            password=password,
            database=database
        )
        self.cursor = self.connection.cursor(dictionary=True)
    def analyze_index_usage(self):
        """分析索引使用情况"""
        query = """
        SELECT 
            OBJECT_SCHEMA,
            OBJECT_NAME,
            INDEX_NAME,
            COUNT_STAR,
            SUM_TIMER_WAIT/1000000000 AS total_wait_seconds
        FROM performance_schema.table_index_usage_statistics
        WHERE INDEX_NAME IS NOT NULL
        ORDER BY COUNT_STAR ASC
        LIMIT 20
        """
        self.cursor.execute(query)
        return self.cursor.fetchall()
    def find_unused_indexes(self, days_threshold=30):
        """查找未使用的索引"""
        query = """
        SELECT 
            t.TABLE_SCHEMA,
            t.TABLE_NAME,
            s.INDEX_NAME,
            GROUP_CONCAT(s.COLUMN_NAME ORDER BY s.SEQ_IN_INDEX) AS columns
        FROM information_schema.TABLES t
        JOIN information_schema.STATISTICS s 
            ON t.TABLE_SCHEMA = s.TABLE_SCHEMA 
            AND t.TABLE_NAME = s.TABLE_NAME
        LEFT JOIN performance_schema.table_index_usage_statistics u
            ON u.OBJECT_SCHEMA = t.TABLE_SCHEMA
            AND u.OBJECT_NAME = t.TABLE_NAME
            AND u.INDEX_NAME = s.INDEX_NAME
        WHERE t.TABLE_SCHEMA NOT IN ('mysql', 'performance_schema', 'information_schema')
          AND t.TABLE_TYPE = 'BASE TABLE'
          AND (u.COUNT_STAR = 0 OR u.COUNT_STAR IS NULL)
        GROUP BY t.TABLE_SCHEMA, t.TABLE_NAME, s.INDEX_NAME
        HAVING COUNT(*) > 1
        """
        self.cursor.execute(query)
        return self.cursor.fetchall()
    def suggest_index_improvements(self):
        """建议索引改进"""
        query = """
        SELECT 
            t.TABLE_SCHEMA,
            t.TABLE_NAME,
            GROUP_CONCAT(DISTINCT c.COLUMN_NAME ORDER BY c.ORDINAL_POSITION) AS foreign_keys,
            COUNT(DISTINCT i.INDEX_NAME) AS existing_indexes
        FROM information_schema.TABLES t
        JOIN information_schema.COLUMNS c 
            ON t.TABLE_SCHEMA = c.TABLE_SCHEMA 
            AND t.TABLE_NAME = c.TABLE_NAME
        LEFT JOIN information_schema.STATISTICS i
            ON i.TABLE_SCHEMA = t.TABLE_SCHEMA
            AND i.TABLE_NAME = t.TABLE_NAME
            AND i.COLUMN_NAME = c.COLUMN_NAME
        WHERE t.TABLE_SCHEMA NOT IN ('mysql', 'performance_schema', 'information_schema')
          AND c.COLUMN_NAME LIKE '%_id'
          AND t.TABLE_TYPE = 'BASE TABLE'
        GROUP BY t.TABLE_SCHEMA, t.TABLE_NAME
        HAVING COUNT(DISTINCT i.INDEX_NAME) < COUNT(DISTINCT c.COLUMN_NAME)
        """
        self.cursor.execute(query)
        return self.cursor.fetchall()
    def generate_optimization_script(self, output_file='index_optimization.sql'):
        """生成优化脚本"""
        unused_indexes = self.find_unused_indexes()
        suggestions = self.suggest_index_improvements()
        with open(output_file, 'w') as f:
            f.write(f"-- Index Optimization Script\n")
            f.write(f"-- Generated: {datetime.now()}\n\n")
            # 删除未使用的索引
            if unused_indexes:
                f.write("-- Drop unused indexes\n")
                for idx in unused_indexes:
                    f.write(f"DROP INDEX {idx['INDEX_NAME']} ON {idx['TABLE_SCHEMA']}.{idx['TABLE_NAME']};\n")
                f.write("\n")
            # 添加建议的索引
            if suggestions:
                f.write("-- Add suggested indexes\n")
                for sug in suggestions:
                    # 简化的索引创建，实际应根据具体列创建
                    f.write(f"CREATE INDEX idx_{sug['TABLE_NAME']}_fk ON {sug['TABLE_SCHEMA']}.{sug['TABLE_NAME']} ({sug['foreign_keys']});\n")
    def close(self):
        self.cursor.close()
        self.connection.close()
# 使用示例
if __name__ == "__main__":
    optimizer = IndexOptimizer(
        host='localhost',
        user='root',
        password='password',
        database='your_database'
    )
    # 分析索引使用情况
    usage = optimizer.analyze_index_usage()
    print("Index Usage Analysis:")
    for row in usage:
        print(f"  {row['OBJECT_NAME']}.{row['INDEX_NAME']}: {row['COUNT_STAR']} uses")
    # 生成优化脚本
    optimizer.generate_optimization_script()
    optimizer.close()

索引维护脚本

定期维护脚本

#!/bin/bash
# index_maintenance.sh
# MySQL索引维护
optimize_mysql_indexes() {
    local DB_HOST=$1
    local DB_USER=$2
    local DB_PASS=$3
    local DB_NAME=$4
    echo "Starting MySQL index optimization..."
    # 获取需要优化的表
    TABLES=$(mysql -h $DB_HOST -u $DB_USER -p$DB_PASS -N -e "
        SELECT CONCAT(TABLE_SCHEMA, '.', TABLE_NAME)
        FROM information_schema.TABLES
        WHERE TABLE_SCHEMA = '$DB_NAME'
          AND TABLE_TYPE = 'BASE TABLE'
          AND DATA_FREE > 10000000  # 碎片大于10MB
    ")
    for table in $TABLES; do
        echo "Optimizing table: $table"
        mysql -h $DB_HOST -u $DB_USER -p$DB_PASS -e "
            ALTER TABLE $table ENGINE=InnoDB;
            ANALYZE TABLE $table;
            OPTIMIZE TABLE $table;
        "
    done
    echo "MySQL index optimization completed"
}
# PostgreSQL索引维护
optimize_postgres_indexes() {
    local DB_HOST=$1
    local DB_PORT=$2
    local DB_NAME=$3
    echo "Starting PostgreSQL index optimization..."
    # 重建碎片化索引
    psql -h $DB_HOST -p $DB_PORT -d $DB_NAME -c "
        SELECT schemaname || '.' || tablename AS table_name,
               indexrelname AS index_name,
               pg_size_pretty(pg_relation_size(indexrelid)) AS size,
               ROUND(100 * (1 - avg_leaf_density)) AS fragmentation_percent
        FROM pgstatindex('all')
        WHERE avg_leaf_density < 0.8
        ORDER BY fragmentation_percent DESC;
    " | while read line; do
        if [[ $line != *"table_name"* ]] && [[ ! -z "$line" ]]; then
            echo "Rebuilding index: $line"
            psql -h $DB_HOST -p $DB_PORT -d $DB_NAME -c "REINDEX INDEX $line;"
        fi
    done
    echo "PostgreSQL index optimization completed"
}
# 定时任务配置
setup_cron_job() {
    # 添加到crontab，每周日凌晨3点执行
    (crontab -l 2>/dev/null; echo "0 3 * * 0 /path/to/index_maintenance.sh mysql") | crontab -
    echo "Cron job configured for weekly index maintenance"
}
# 主函数
main() {
    case "$1" in
        mysql)
            optimize_mysql_indexes "localhost" "root" "password" "your_database"
            ;;
        postgres)
            optimize_postgres_indexes "localhost" "5432" "your_database"
            ;;
        setup_cron)
            setup_cron_job
            ;;
        *)
            echo "Usage: $0 {mysql|postgres|setup_cron}"
            exit 1
            ;;
    esac
}
main "$@"

最佳实践建议

索引优化原则

1. 定期检查：每周检查索引使用情况
2. 按需创建：只创建必要的索引，避免过度索引
3. 监控性能：监控慢查询，根据查询模式调整索引
4. 及时清理：定期删除未使用或低效索引

注意事项

• 在生产环境操作前，先在测试环境验证
• 创建或删除索引时注意锁定影响
• 对大表的索引操作选择低峰期
• 备份重要数据和索引定义

这些脚本可以帮助你系统化地管理和优化数据库索引,提高查询性能和数据库整体效率。