如何基于NumPy数组实现MySQL表的快速批量更新

性能慢根因 #

逐行执行单条UPDATE语句的开销来自每次都要经历SQL解析、事务提交、网络往返开销，6000行数据的累计开销会被放大数倍，以下是两种高效实现方案：

方案1：单条CASE WHEN批量更新（适合万行以内数据量最优） #

将所有更新逻辑拼接为单条SQL语句，仅执行1次数据库请求即可完成全部更新：

import numpy as np
import pymysql

# 示例numpy数组
x = np.array([['a', '2'], ['b', '2'] , ['c', '1']])

when_clauses = []
params = []
for name, val in x:
    when_clauses.append("WHEN %s THEN %s")
    params.extend([name, val])

# 收集所有匹配的name，避免误更新不相关行
name_list = x[:, 0].tolist()
params.extend(name_list)

sql = f"""
UPDATE test_table
SET value = CASE Name
    {' '.join(when_clauses)}
    ELSE value
END
WHERE Name IN ({','.join(['%s']*len(name_list))})
"""

# 数据库连接执行
conn = pymysql.connect(
    host="你的数据库地址",
    user="你的用户名",
    password="你的密码",
    database="你的数据库名",
    charset="utf8mb4"
)
cursor = conn.cursor()
cursor.execute(sql, params)
conn.commit()
cursor.close()
conn.close()

*该方案采用参数化查询，无需手动处理字符串转义，可避免SQL注入风险。

方案2：临时表关联更新（适合十万行以上大数据量） #

如果后续数据量持续增长，用临时表方案性能更稳定，不会出现SQL语句过长的问题：

import numpy as np
import pymysql

x = np.array([['a', '2'], ['b', '2'] , ['c', '1']])
# numpy转列表方便批量插入
data = x.tolist()

conn = pymysql.connect(
    host="你的数据库地址",
    user="你的用户名",
    password="你的密码",
    database="你的数据库名",
    charset="utf8mb4"
)
cursor = conn.cursor()

# 创建临时表，会话结束后自动销毁
cursor.execute("CREATE TEMPORARY TABLE tmp_update (name VARCHAR(255) PRIMARY KEY, val VARCHAR(255)) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4")

# 批量插入数据到临时表
cursor.executemany("INSERT INTO tmp_update (name, val) VALUES (%s, %s)", data)

# 关联更新正式表
cursor.execute("""
UPDATE test_table t
INNER JOIN tmp_update tmp ON t.Name = tmp.name
SET t.value = tmp.val
""")

conn.commit()
cursor.close()
conn.close()

补充优化建议 #

• 提前为test_table的Name字段建立索引，可大幅提升匹配速度，若已存在索引可忽略。
• 两种方案的性能均远高于逐行更新，6000行数据实测耗时可从秒级降到百毫秒级。

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