Python 数据库优化：SQL 索引与查询优化

Python 应用连接数据库时，性能瓶颈往往不在代码逻辑，而在 SQL 查询效率。创建合适的索引 和 重写低效查询 是提升响应速度最直接有效的方法。以下步骤适用于 SQLite、MySQL、PostgreSQL 等主流关系型数据库。

诊断慢查询

在优化前，必须先定位问题所在。

1. 启用数据库慢查询日志。以 MySQL 为例，在配置文件 my.cnf 中添加：

   slow_query_log = 1
   slow_query_log_file = /var/log/mysql/slow.log
   long_query_time = 1

   表示记录执行时间超过 1 秒的 SQL。

2. 使用 EXPLAIN 分析查询计划。在 Python 中执行带 EXPLAIN 前缀的 SQL：

   cursor.execute("EXPLAIN SELECT * FROM users WHERE email = %s", (email,))
   print(cursor.fetchall())

   关注输出中的 type（访问类型）和 rows（扫描行数）。若 type 为 ALL 且 rows 接近全表行数，说明未走索引。

3. 测量实际执行时间。在 Python 中用 time.time() 包裹查询：

   import time
   start = time.time()
   cursor.execute("SELECT ...")
   elapsed = time.time() - start
   print(f"Query took {elapsed:.4f} seconds")

创建高效索引

索引不是越多越好，而是要精准覆盖高频查询条件。

1. 为 WHERE 子句中的列建索引。例如查询用户邮箱：

   CREATE INDEX idx_users_email ON users(email);

2. 组合索引遵循最左前缀原则。若常按 (status, created_at) 联合筛选：

   CREATE INDEX idx_users_status_created ON users(status, created_at);

   此索引可加速 WHERE status = 'active' 和 WHERE status = 'active' AND created_at > '2023-01-01'，但无法加速仅 WHERE created_at > '2023-01-01' 的查询。

3. 避免在索引列上使用函数。以下写法会使索引失效：

   -- 错误：对列使用函数
   SELECT * FROM orders WHERE YEAR(created_at) = 2023;
   
   -- 正确：改写为范围查询
   SELECT * FROM orders 
   WHERE created_at >= '2023-01-01' AND created_at < '2024-01-01';

4. 定期检查索引使用率。在 MySQL 中运行：

   SELECT * FROM sys.schema_unused_indexes;

   删除长期未使用的索引以减少写入开销。

重写低效查询

即使有索引，糟糕的 SQL 写法仍会导致全表扫描。

1. 用 JOIN 替代子查询。子查询可能被多次执行：

   -- 低效：相关子查询
   SELECT name FROM users 
   WHERE id IN (SELECT user_id FROM orders WHERE amount > 100);
   
   -- 高效：JOIN
   SELECT DISTINCT u.name 
   FROM users u
   JOIN orders o ON u.id = o.user_id 
   WHERE o.amount > 100;

2. 限制返回字段数量。避免 SELECT *，只取必要列：

   # 错误：获取所有字段
   cursor.execute("SELECT * FROM products WHERE category = %s", (cat,))
   
   # 正确：指定字段
   cursor.execute("SELECT id, name, price FROM products WHERE category = %s", (cat,))

3. 分页查询避免 OFFSET 深度过大。传统分页在万级数据后急剧变慢：

   -- 低效：跳过 10000 行
   SELECT * FROM logs ORDER BY id LIMIT 20 OFFSET 10000;
   
   -- 高效：基于上一页最大 ID
   SELECT * FROM logs WHERE id > 10000 ORDER BY id LIMIT 20;

4. 聚合查询预计算。对实时性要求不高的统计，用物化视图或定时任务：

   # 每小时更新一次统计表
   cursor.execute("""
       INSERT INTO daily_sales_summary (date, total)
       SELECT CURDATE(), SUM(amount) 
       FROM orders 
       WHERE DATE(created_at) = CURDATE()
       ON DUPLICATE KEY UPDATE total = VALUES(total)
   """)

Python 代码层配合优化

数据库优化需与应用代码协同。

1. 使用参数化查询防止 SQL 注入，同时让数据库复用执行计划：

   # 正确：参数化
   cursor.execute("SELECT * FROM items WHERE category = %s", (category,))
   
   # 错误：字符串拼接
   cursor.execute(f"SELECT * FROM items WHERE category = '{category}'")

2. 批量操作替代循环单条执行。插入 1000 条记录时：

   # 低效：逐条插入
   for item in items:
       cursor.execute("INSERT INTO logs (msg) VALUES (%s)", (item,))
   
   # 高效：批量插入
   cursor.executemany("INSERT INTO logs (msg) VALUES (%s)", items)

3. 显式管理事务。避免自动提交导致频繁磁盘 I/O：

   conn.autocommit = False
   try:
       for record in data:
           cursor.execute("UPDATE accounts SET balance = balance - %s WHERE id = %s", 
                         (amount, user_id))
       conn.commit()
   except Exception:
       conn.rollback()

4. 连接池复用数据库连接。使用 DBUtils 或 SQLAlchemy 连接池：

   from dbutils.pooled_db import PooledDB
   
   pool = PooledDB(
       creator=pymysql,
       maxconnections=20,
       host='localhost',
       user='user',
       password='pass',
       database='app_db'
   )
   conn = pool.connection()  # 从池中获取连接

监控与持续优化

优化是持续过程，需建立反馈机制。

1. 记录关键查询的执行时间。在 Python 日志中添加耗时信息：

   import logging
   logger = logging.getLogger(__name__)
   
   def timed_query(sql, params):
       start = time.perf_counter()
       cursor.execute(sql, params)
       duration = (time.perf_counter() - start) * 1000
       if duration > 100:  # 超过100ms记录警告
           logger.warning(f"Slow query ({duration:.1f}ms): {sql}")
       return cursor.fetchall()

2. 定期分析表统计信息。数据库依赖统计信息生成执行计划：

   -- MySQL
   ANALYZE TABLE users;
   
   -- PostgreSQL
   ANALYZE users;

3. 设置查询超时。防止长查询拖垮整个服务：

   # MySQL 设置语句超时10秒
   cursor.execute("SET SESSION max_execution_time = 10000")
   cursor.execute("SELECT ...")

4. 对比优化前后指标。用相同数据集测试：

   • 查询平均响应时间
   • CPU 和 I/O 使用率
   • 每秒查询数（QPS）

删除无效索引
重写含函数的 WHERE 条件
用游标分页替代 OFFSET