如何加速MySQL数据更新？现有循环代码优化及NumPy可行性探讨

原代码的性能瓶颈 #

你的代码速度慢核心是这几个问题：

• 每次循环都新建cursor：完全没必要，cursor可以复用，重复创建只会增加额外开销
• 每更新一条就commit一次：commit是磁盘IO操作，频繁提交会拖慢整个流程，应该攒一批再提交
• 逐行执行UPDATE：单条UPDATE的网络往返、数据库解析执行成本都很高，批量处理能大幅降低这些开销

是否需要修改循环逻辑？ #

必须改，先把最基础的优化做了：

• 把cursor创建移到循环外面
• 关闭自动提交，批量提交（比如每100条提交一次，最后再提交剩余的）
• 去掉多余的currency变量，直接用i里的元素，减少内存操作

优化后的基础版本代码：

try:
    g = 1
    batch_size = 100  # 批量提交的大小，可根据数据量调整
    count = 0
    with connection.cursor() as cursor:
        connection.autocommit(False)  # 关闭自动提交，手动控制事务
        insert_query = "UPDATE gate SET symbol = %s, bidPX = %s, askPx = %s WHERE id = %s"
        for i in gate_io().values.tolist():
            if i[1] != 0 and i[1] != '':
                # 直接传参数，不用额外转存到currency
                cursor.execute(insert_query, (i[0], i[1], i[2], g))
                count += 1
                g += 1
                # 达到批量大小就提交一次
                if count % batch_size == 0:
                    connection.commit()
            else:
                continue
        # 提交剩余未提交的数据
        if count % batch_size != 0:
            connection.commit()
finally:
    connection.close()

能用NumPy实现吗？ #

NumPy本身不直接操作数据库，但可以用它来高效整理数据，减少Python循环里的操作开销。比如先把gate_io()返回的DataFrame转成NumPy数组，再批量提取有效数据，然后传给cursor批量执行：

import numpy as np

try:
    g = 1
    batch_size = 100
    count = 0
    # 转成NumPy数组，比列表操作更快
    data = gate_io().values
    with connection.cursor() as cursor:
        connection.autocommit(False)
        insert_query = "UPDATE gate SET symbol = %s, bidPX = %s, askPx = %s WHERE id = %s"
        for row in data:
            if row[1] != 0 and row[1] != '':
                cursor.execute(insert_query, (row[0], row[1], row[2], g))
                count +=1
                g +=1
                if count % batch_size ==0:
                    connection.commit()
        if count % batch_size !=0:
            connection.commit()
finally:
    connection.close()

注意：这里NumPy主要是提升数据遍历的效率，核心加速还是靠批量提交和减少cursor创建次数。

其他更高效的方案 #

1. 用CASE WHEN合并成单条UPDATE语句

把多条更新合并成一条SQL，减少网络往返次数，比如：

try:
    update_clauses = []
    params = []
    g =1
    for i in gate_io().values.tolist():
        if i[1] !=0 and i[1] != '':
            # 构造CASE WHEN的子句
            update_clauses.append("WHEN id = %s THEN %s")
            params.extend([g, i[0]])
            update_clauses.append("WHEN id = %s THEN %s")
            params.extend([g, i[1]])
            update_clauses.append("WHEN id = %s THEN %s")
            params.extend([g, i[2]])
            g +=1
    if update_clauses:
        # 拼接成完整的UPDATE语句
        update_query = f"""
            UPDATE gate 
            SET 
                symbol = CASE {' '.join(update_clauses[::3])} ELSE symbol END,
                bidPX = CASE {' '.join(update_clauses[1::3])} ELSE bidPX END,
                askPx = CASE {' '.join(update_clauses[2::3])} ELSE askPx END
            WHERE id IN ({', '.join(['%s']*(g-1))})
        """
        # 加入id的过滤参数
        params.extend(list(range(1, g)))
        with connection.cursor() as cursor:
            cursor.execute(update_query, params)
            connection.commit()
finally:
    connection.close()

这条SQL会一次性处理所有符合条件的更新，大幅减少网络交互。

2. 使用INSERT ... ON DUPLICATE KEY UPDATE

如果id是表的主键或唯一键，可以把更新操作转成批量插入+冲突更新，MySQL对批量插入的优化更好：

try:
    batch_data = []
    g =1
    for i in gate_io().values.tolist():
        if i[1] !=0 and i[1] != '':
            # 构造插入的参数（id, symbol, bidPX, askPx）
            batch_data.append((g, i[0], i[1], i[2]))
            g +=1
    if batch_data:
        insert_query = """
            INSERT INTO gate (id, symbol, bidPX, askPx)
            VALUES (%s, %s, %s, %s)
            ON DUPLICATE KEY UPDATE
            symbol = VALUES(symbol),
            bidPX = VALUES(bidPX),
            askPx = VALUES(askPx)
        """
        with connection.cursor() as cursor:
            # executemany批量执行
            cursor.executemany(insert_query, batch_data)
            connection.commit()
finally:
    connection.close()

executemany会把批量参数打包发送给数据库，比逐行执行快很多，加上ON DUPLICATE KEY UPDATE的逻辑，效果比逐行UPDATE好太多。

3. 使用LOAD DATA INFILE（超大量数据首选）

如果你的数据量特别大（比如几十万条以上），可以把数据导出成CSV文件，然后用MySQL的LOAD DATA INFILE命令导入，这是MySQL最快的批量数据写入方式：

import pandas as pd

try:
    # 先把有效数据导出成CSV
    df = gate_io()
    df = df[(df.iloc[:,1] !=0) & (df.iloc[:,1] != '')]
    df['id'] = range(1, len(df)+1)  # 生成id列
    df[['id', 'symbol', 'bidPX', 'askPx']].to_csv('update_data.csv', index=False, header=False)
    
    # 执行LOAD DATA命令
    load_query = """
        LOAD DATA INFILE '/path/to/update_data.csv'
        INTO TABLE gate
        FIELDS TERMINATED BY ','
        LINES TERMINATED BY '\n'
        (id, symbol, bidPX, askPx)
        ON DUPLICATE KEY UPDATE
        symbol = VALUES(symbol),
        bidPX = VALUES(bidPX),
        askPx = VALUES(askPx)
    """
    with connection.cursor() as cursor:
        cursor.execute(load_query)
        connection.commit()
finally:
    connection.close()

注意：需要确保MySQL服务有权限读取这个CSV文件，或者用LOAD DATA LOCAL INFILE（需要开启客户端的local_infile参数）。

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