如何加速Python脚本更新SQL Orders表的State与Country列？

问题背景 #

我有一个SQL的Orders表，包含StoreOrderNumber、State和Country字段，其中State和Country为空白列，需要用以下DataFrame数据填充：

data = {
    'StoreOrderNumber': [1001, 1002, 1003, 1004, 1005, 1006, 1007, 1008, 1009, 1010],
    'State': ['CA', 'NY', 'TX', 'IL', 'FL', 'PA', 'OH', 'GA', 'MI', 'NC'],
    'Country': ['USA', 'USA', 'USA', 'USA', 'USA', 'USA', 'USA', 'USA', 'USA', 'USA']
}

当前使用的Python代码运行速度极慢，每秒仅能完成约7次更新：

# Addin data to Orders table
sn_conn = DatabaseConnection(config_file)
sn_conn.sqldb(env = env)
for each in tqdm(orders_final.StoreOrderNumber):
    state= orders_final.loc[orders_final.StoreOrderNumber == each, 'State'].values[0]
    country = orders_final.loc[orders_final.StoreOrderNumber == each, 'Country'].values[0]
    sn_conn.engine.execute(f"""UPDATE Orders SET State = '{state}' WHERE StoreOrderNumber = '{each}' """)
    sn_conn.engine.execute(f"""UPDATE Orders SET Country = '{country}' WHERE StoreOrderNumber = '{each}' """)
sn_conn.close()

核心优化思路 #

原代码的问题在于单条循环+多次数据库交互，每次更新都要建立会话、执行SQL、返回结果，开销极大。优化方向是减少数据库交互次数，利用数据库的批量处理能力。

方案1：合并单条更新语句 #

把原本两次UPDATE合并成一次，直接更新两个字段，减少一半的数据库交互；同时提前构建DataFrame映射字典，避免循环内重复查询DataFrame：

# 提前构建映射字典，减少循环内的DataFrame查询开销
state_map = orders_final.set_index('StoreOrderNumber')['State'].to_dict()
country_map = orders_final.set_index('StoreOrderNumber')['Country'].to_dict()

sn_conn = DatabaseConnection(config_file)
sn_conn.sqldb(env = env)
for order_num in tqdm(orders_final.StoreOrderNumber):
    state = state_map[order_num]
    country = country_map[order_num]
    # 合并两个字段的更新到同一条SQL
    sn_conn.engine.execute(f"""
        UPDATE Orders 
        SET State = '{state}', Country = '{country}' 
        WHERE StoreOrderNumber = '{order_num}'
    """)
sn_conn.close()

方案2：CASE WHEN批量更新（推荐中小数据量） #

生成一条包含所有更新逻辑的SQL语句，只和数据库交互一次，速度会有数量级提升：

sn_conn = DatabaseConnection(config_file)
sn_conn.sqldb(env = env)

# 构建State字段的CASE WHEN逻辑
state_case_clauses = [
    f"WHEN StoreOrderNumber = {row['StoreOrderNumber']} THEN '{row['State']}'"
    for _, row in orders_final.iterrows()
]
# 构建Country字段的CASE WHEN逻辑
country_case_clauses = [
    f"WHEN StoreOrderNumber = {row['StoreOrderNumber']} THEN '{row['Country']}'"
    for _, row in orders_final.iterrows()
]

# 拼接完整的批量更新SQL
batch_update_sql = f"""
    UPDATE Orders
    SET 
        State = CASE {' '.join(state_case_clauses)} ELSE State END,
        Country = CASE {' '.join(country_case_clauses)} ELSE Country END
    WHERE StoreOrderNumber IN ({','.join(map(str, orders_final['StoreOrderNumber'].tolist()))})
"""

# 执行一次批量更新
sn_conn.engine.execute(batch_update_sql)
sn_conn.close()

注意：如果数据量超过10万条，生成的SQL可能过长，此时建议用临时表方案。

方案3：临时表+JOIN更新（推荐大数据量） #

先把DataFrame数据导入数据库临时表，再通过JOIN完成批量更新，适合几十万条以上的数据：

sn_conn = DatabaseConnection(config_file)
sn_conn.sqldb(env = env)

# 将DataFrame写入临时表（不同数据库临时表语法有差异，以下是SQL Server示例）
orders_final.to_sql(
    name='#temp_order_updates',
    con=sn_conn.engine,
    if_exists='replace',
    index=False
)

# 通过JOIN执行高效更新
update_sql = """
    UPDATE o
    SET o.State = t.State, o.Country = t.Country
    FROM Orders o
    JOIN #temp_order_updates t 
        ON o.StoreOrderNumber = t.StoreOrderNumber
"""

sn_conn.engine.execute(update_sql)
sn_conn.close()

MySQL用户可以把临时表名改为temp_order_updates，并添加temp_table=True参数到to_sql方法。

关键额外优化 #

• 确保Orders表的StoreOrderNumber字段创建了唯一索引，这能让WHERE和JOIN语句的执行速度提升数倍
• 避免字符串拼接SQL时的注入风险，建议用参数化查询（比如SQLAlchemy的绑定参数），示例：

  # 参数化更新示例（替换方案1中的字符串拼接）
  sn_conn.engine.execute(
      "UPDATE Orders SET State = :state, Country = :country WHERE StoreOrderNumber = :order_num",
      {"state": state, "country": country, "order_num": order_num}
  )
  

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