使用python-mysql-connector批量更新MySQL数据提速咨询

问题背景 #

我有一个包含约10000个唯一字符串的id列表inactive_ids，需要将MySQL表test_table中NoticeId属于该列表的所有行的Active字段更新为'No'。目前使用mysql.connector通过循环逐个执行UPDATE语句，每次迭代耗时约0.7秒，处理10000条数据总计需约2小时，急需更高效的实现方式。

当前代码 #

# inactive_ids are unique strings something like shown below
# inactive_ids = ['a9okeoko', 'sdfhreaa', 'xsdfasy', ..., 'asdfad']

# initialize connection
mydb = mysql.connector.connect(
    user="REMOVED", 
    password="REMOVED", 
    host="REMOVED", 
    database="REMOVED"
)

# initialize cursor
mycursor = mydb.cursor(buffered=True)

# Function to execute multiple lines
def alter(state, msg, count):
    result = mycursor.execute(state, multi=True)
    result.send(None)
    print(str(count), ': ', msg, result)
    count += 1
    return count

# Try to execute, throw exception if fails
try:
    count = 0
    for Id in inactive_ids:
        
        # SAVE THE QUERY AS STRING
        sql_update = "UPDATE test_table SET Active = 'No' WHERE NoticeId = '" + Id + "'"
    
        # ALTER
        count = alter(sql_update, "done", count)

    # commits all changes to the database
    mydb.commit()
except Exception as e:
    mydb.rollback()
    raise e

优化方案 #

方案1：分批使用IN子句（最简便高效） #

MySQL的IN子句支持一次性传入多个值，但默认有长度限制（通常是1000个元素），所以可以将inactive_ids分成若干批次（比如每1000个一组），每次执行一条批量UPDATE语句，大幅减少数据库交互次数。同时使用参数化查询避免SQL注入风险，也提升执行效率。

示例代码：

mydb = mysql.connector.connect(
    user="REMOVED", 
    password="REMOVED", 
    host="REMOVED", 
    database="REMOVED"
)
mycursor = mydb.cursor()

# 分批大小，根据MySQL配置调整，建议不超过1000
batch_size = 1000

try:
    # 将列表分批处理
    for i in range(0, len(inactive_ids), batch_size):
        batch = inactive_ids[i:i+batch_size]
        # 生成对应数量的占位符
        placeholders = ', '.join(['%s'] * len(batch))
        sql = f"UPDATE test_table SET Active = 'No' WHERE NoticeId IN ({placeholders})"
        mycursor.execute(sql, batch)
        print(f"已处理第{i//batch_size + 1}批，共{len(batch)}条数据")
    
    mydb.commit()
    print("所有更新操作完成")
except Exception as e:
    mydb.rollback()
    raise e
finally:
    mycursor.close()
    mydb.close()

方案2：使用临时表JOIN更新（超大量数据场景） #

如果id数量远超10000，或者表数据量极大，可以先将所有id插入临时表，再通过JOIN方式执行更新，这种方式在数据量极大时性能更优。

示例代码：

mydb = mysql.connector.connect(
    user="REMOVED", 
    password="REMOVED", 
    host="REMOVED", 
    database="REMOVED"
)
mycursor = mydb.cursor()

try:
    # 创建临时表
    mycursor.execute("CREATE TEMPORARY TABLE temp_inactive_ids (notice_id VARCHAR(255) PRIMARY KEY)")
    
    # 批量插入id到临时表
    insert_sql = "INSERT INTO temp_inactive_ids (notice_id) VALUES (%s)"
    mycursor.executemany(insert_sql, [(id,) for id in inactive_ids])
    
    # 通过JOIN执行更新
    update_sql = """
        UPDATE test_table t
        JOIN temp_inactive_ids tmp ON t.NoticeId = tmp.notice_id
        SET t.Active = 'No'
    """
    mycursor.execute(update_sql)
    
    mydb.commit()
    print("所有更新操作完成")
except Exception as e:
    mydb.rollback()
    raise e
finally:
    mycursor.close()
    mydb.close()

额外优化点 #

• 确保test_table的NoticeId字段创建了索引，这会极大加速WHERE和JOIN的查询效率
• 原代码中使用字符串拼接SQL存在严重的SQL注入风险，优化方案均采用参数化查询避免此问题
• 关闭不必要的buffered=True（除非有特殊需求），减少内存消耗

内容的提问来源于stack exchange，提问作者Joey O'Neill