如何加速含SQL查询的代码？65万行DataFrame循环查询耗时过长

嘿，这个问题我太有共鸣了——逐行查数据库简直是性能杀手！65万次查询的连接开销、网络传输成本加起来，直接把时间拉到了25小时，完全没必要在并行循环上死磕，咱们换个思路，让数据库发挥它的批量处理优势，效率能提升好几个数量级。

先戳破核心问题：循环逐行发起SQL查询是最大的性能瓶颈，哪怕用joblib做并行，也只是把65万次查询分给多个进程，数据库的单查询开销依然存在，甚至可能因为并发连接过多变慢。下面是两个最优解决方案：

解决方案1：批量导入+关联查询（效率最高） #

这是我最推荐的方式，把所有查询条件一次性传给数据库，让数据库帮你完成所有统计，最后只需要一次把结果拉回合并。

步骤1：给原DataFrame加唯一标识，导出关键参数 #

先给你的table加一个row_id（用原索引就行），然后把需要的查询参数（good, store, start, row_id）导入到数据库的临时表，比如叫temp_query_params：

# 给原表加行标识，后续用来匹配结果
table['row_id'] = table.index

# 导出关键列到数据库临时表
# 不同数据库的临时表语法略有差异，比如MySQL用CREATE TEMPORARY TABLE，PostgreSQL用CREATE TEMP TABLE
table[[table.columns[0], table.columns[1], table.columns[6], 'row_id']].to_sql(
    'temp_query_params', 
    connection, 
    if_exists='replace', 
    index=False
)

步骤2：写关联SQL一次性计算所有统计值 #

直接在数据库里关联临时表和my_table，一次性算出所有行的统计结果：

SELECT 
    t.row_id,
    AVG(m.sale) AS avg_sale,
    SUM(m.sale) AS sum_sale,
    MAX(m.sale) AS max_sale,
    MIN(m.sale) AS min_sale
FROM temp_query_params t
LEFT JOIN my_table m 
    ON m.good_id = t.{good_col_name}  -- 替换成你原表第0列的实际列名
    AND m.store_id = t.{store_col_name} -- 替换成原表第1列的实际列名
    AND m.date_id BETWEEN DATEADD(MONTH, -2, t.{start_col_name}) 
        AND DATEADD(MONTH, -1, t.{start_col_name}) -- 替换成原表第6列的实际列名
GROUP BY t.row_id

步骤3：拉回结果合并到原表 #

把查询结果读回Pandas，用row_id匹配回原表，填充对应的列：

# 执行查询
stats_result = pd.read_sql(上面的SQL语句, connection)

# 合并回原表
table = table.merge(stats_result, on='row_id', how='left')

# 把统计值填充到指定列
table.iloc[:, 13] = table['avg_sale']
table.iloc[:, 14] = table['sum_sale']
table.iloc[:, 15] = table['max_sale']
table.iloc[:, 16] = table['min_sale']

这个方法全程只需要2次数据库交互，耗时会从25小时直接降到几分钟，完全取决于数据库的计算能力。

解决方案2：分块构造批量查询（无临时表权限时用） #

如果没有创建临时表的权限，那就把查询条件分成若干块（比如每块1000条），构造包含多个条件的SQL，减少查询次数：

import pandas as pd
from tqdm import tqdm_notebook

# 给原表加行标识
table['row_id'] = table.index
chunk_size = 1000  # 每块的大小，根据数据库支持调整
all_results = []

for i in tqdm_notebook(range(0, len(table), chunk_size)):
    chunk = table.iloc[i:i+chunk_size]
    # 构造参数化查询的条件和参数列表
    conditions = []
    params = []
    for _, row in chunk.iterrows():
        good = row[0]
        store = row[1]
        start = row[6]
        # 用占位符避免SQL注入，同时构造条件
        conditions.append("(good_id = %s AND store_id = %s AND date_id BETWEEN DATEADD(MONTH, -2, %s) AND DATEADD(MONTH, -1, %s))")
        params.extend([good, store, start, start])
    
    # 合并条件，执行查询
    where_clause = " OR ".join(conditions)
    query = f"""
        SELECT good_id, store_id, %s as start_date,
               AVG(sale) AS avg_sale, SUM(sale) AS sum_sale,
               MAX(sale) AS max_sale, MIN(sale) AS min_sale
        FROM my_table
        WHERE {where_clause}
        GROUP BY good_id, store_id, start_date
    """
    temp = pd.read_sql(query, connection, params=params)
    all_results.append(temp)

# 合并所有结果，关联回原表
stats_result = pd.concat(all_results)
table = table.merge(
    stats_result,
    left_on=[table.columns[0], table.columns[1], table.columns[6]],
    right_on=['good_id', 'store_id', 'start_date'],
    how='left'
)

# 填充统计列
table.iloc[:,13:17] = table[['avg_sale', 'sum_sale', 'max_sale', 'min_sale']].values

注意：这里一定要用参数化查询（params参数），绝对不要直接字符串拼接变量，避免SQL注入风险。

为什么不推荐joblib/numba？ #

• numba：它主要是加速Python的数值计算代码，对数据库查询这种IO密集型操作完全无效，因为numba无法优化网络请求和数据库交互的部分。
• joblib：并行确实能减少一点时间，但本质还是在做65万次查询，数据库的单查询开销依然存在，最多只能把时间降到几分之一（比如8核降到3小时），远不如批量查询高效。

内容的提问来源于stack exchange，提问作者Fissium