Python批量删除SQL记录耗时过长，如何优化性能？

DevOps 小满

2026-6-19

批量删除SQL记录的性能优化方案

问题描述

我有一个包含8000个唯一CONVERSATIONID的delete_ids变量，原本通过循环逐个从SQL表中删除这些ID，还用try-except跳过删除失败的ID，但整个过程耗时近40分钟，而且delete_ids的数量会变动。有没有更高效的性能优化方法？

现有代码

failed_deletions = []

if len(matching_records_sql) < len(matching_records_df):
    delete_ids = matching_records_sql['CONVERSATIONID'].tolist()
    if delete_ids: 
        delete_query = "DELETE FROM MY_TABLE WHERE CONVERSATIONID = ?"
        
        conn = pyodbc.connect(r'DRIVER={ODBC Driver 17 for SQL Server};'
                              r'SERVER=My_Server_Name;'
                              r'DATABASE=Database_Name;'
                              r'Trusted_Connection=yes;')
        cursor = conn.cursor()
        
        for id in delete_ids:
            try:
                cursor.execute(delete_query, (id,))
                conn.commit()
            except Exception as e:
                print(f"Failed to delete CONVERSATIONID {id}: {e}")
                failed_deletions.append(id)
        
        #conn.close()
        print('List of deleted ids from SQL table: ', [id for id in delete_ids if id not in failed_deletions])
        print('List of IDs that failed to delete: ', failed_deletions)
    else:
        print('No IDs to delete.')
else:
    print('matching_records_sql is not < matching_records_df')

优化方案

1. 批量删除（核心优化）

循环逐个删除的最大问题是网络往返次数过多，8000次请求意味着8000次和数据库的交互。改成批量删除，把多个ID打包成一次SQL操作，能大幅减少耗时。注意SQL Server的IN子句默认最多支持2100个参数，所以要分批次处理。

示例代码：

failed_deletions = []
batch_size = 1000  # 每批次处理1000个ID，避开参数数量限制

if len(matching_records_sql) < len(matching_records_df):
    delete_ids = matching_records_sql['CONVERSATIONID'].tolist()
    if delete_ids:
        conn = pyodbc.connect(r'DRIVER={ODBC Driver 17 for SQL Server};'
                              r'SERVER=My_Server_Name;'
                              r'DATABASE=Database_Name;'
                              r'Trusted_Connection=yes;')
        cursor = conn.cursor()

        # 分批次处理ID
        for idx in range(0, len(delete_ids), batch_size):
            batch_ids = delete_ids[idx:idx+batch_size]
            # 生成对应数量的占位符
            placeholders = ', '.join(['?' for _ in batch_ids])
            delete_query = f"DELETE FROM MY_TABLE WHERE CONVERSATIONID IN ({placeholders})"
            
            try:
                cursor.execute(delete_query, batch_ids)
                conn.commit()
                
                # 校验当前批次中未被删除的ID（用于统计失败记录）
                check_query = f"SELECT CONVERSATIONID FROM MY_TABLE WHERE CONVERSATIONID IN ({placeholders})"
                cursor.execute(check_query, batch_ids)
                remaining_ids = [row[0] for row in cursor.fetchall()]
                failed_deletions.extend(remaining_ids)
            except Exception as e:
                print(f"批次删除失败（起始索引{idx}）: {e}")
                # 批次失败时降级为逐个处理该批次ID
                for single_id in batch_ids:
                    try:
                        cursor.execute("DELETE FROM MY_TABLE WHERE CONVERSATIONID = ?", (single_id,))
                        conn.commit()
                    except Exception as e_single:
                        print(f"删除ID {single_id}失败: {e_single}")
                        failed_deletions.append(single_id)
        
        conn.close()
        deleted_ids = [id for id in delete_ids if id not in failed_deletions]
        print('已删除的SQL记录ID列表: ', deleted_ids)
        print('删除失败的ID列表: ', failed_deletions)
    else:
        print('没有需要删除的ID。')
else:
    print('matching_records_sql的数量不小于matching_records_df')

2. 减少事务提交次数

原代码每次循环都调用conn.commit()，这会产生大量事务日志开销。改成每批次提交一次，甚至可以在所有批次处理完成后一次性提交（但要注意事务规模，避免长时间锁表影响其他业务）。

3. 预校验无效ID

删除前先查询哪些ID在表中不存在，直接标记为失败，不用尝试删除，减少无效的SQL操作：

# 预查询存在的ID
placeholders = ', '.join(['?' for _ in delete_ids])
check_query = f"SELECT CONVERSATIONID FROM MY_TABLE WHERE CONVERSATIONID IN ({placeholders})"
cursor.execute(check_query, delete_ids)
existing_ids = [row[0] for row in cursor.fetchall()]
# 直接标记不存在的ID为删除失败
non_existing_ids = [id for id in delete_ids if id not in existing_ids]
failed_deletions.extend(non_existing_ids)

# 只对存在的ID执行批量删除
if existing_ids:
    for idx in range(0, len(existing_ids), batch_size):
        batch = existing_ids[idx:idx+batch_size]
        placeholders_batch = ', '.join(['?' for _ in batch])
        delete_query = f"DELETE FROM MY_TABLE WHERE CONVERSATIONID IN ({placeholders_batch})"
        cursor.execute(delete_query, batch)
        conn.commit()

4. 其他辅助优化

索引优化：确保MY_TABLE的CONVERSATIONID字段创建了索引，这样删除时的WHERE子句查询会更快，减少数据库扫描时间。
连接池：使用连接池复用数据库连接，避免频繁创建和关闭连接的开销。
驱动版本：确保使用最新的ODBC驱动版本，新版本通常有性能优化。

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