如何通过pyodbc提升数据库更新与插入操作的执行速度

方案1：利用数据库原生UPSERT语句 + 批量执行 #

大部分主流数据库（PostgreSQL、MySQL、SQL Server）都支持原生的UPSERT语法，直接在数据库层面完成“存在则更新，不存在则插入”的逻辑，配合批量数据传入，能大幅减少循环执行的开销。

以PostgreSQL为例，使用execute_values批量传入数据，结合ON CONFLICT语法：

import psycopg2
from psycopg2.extras import execute_values
from psycopg2 import sql

# 从API获取的JSON数据，转成列表字典格式
api_data = [{"item_id": 1, "name": "test_update"}, {"item_id": 3, "name": "test_insert"}]

# 数据库连接
conn = psycopg2.connect("dbname=your_db user=your_user password=your_pass")
cur = conn.cursor()

# 构造字段列表和UPSERT语句
fields = api_data[0].keys()
insert_stmt = sql.SQL("""
    INSERT INTO your_table ({fields})
    VALUES %s
    ON CONFLICT (item_id) DO UPDATE SET
        {updates}
""").format(
    fields=sql.SQL(', ').join(map(sql.Identifier, fields)),
    updates=sql.SQL(', ').join([
        sql.SQL("{field} = EXCLUDED.{field}").format(field=sql.Identifier(field))
        for field in fields if field != 'item_id'
    ])
)

# 批量执行
execute_values(cur, insert_stmt, [tuple(d.values()) for d in api_data])

conn.commit()
cur.close()
conn.close()

MySQL则可以用ON DUPLICATE KEY UPDATE语法，配合executemany（或df.to_sql的multi方法）实现批量UPSERT。

方案2：先批量查询已存在ID，拆分插入/更新批次 #

如果不想依赖数据库原生UPSERT，可以先批量筛选出已存在的item_id，再分别处理插入和更新：

1. 从API数据中提取所有item_id，批量查询数据库中已存在的ID
2. 将数据拆分为「待插入」和「待更新」两个子集
3. 对插入子集用df.to_sql(method='multi', chunksize=500)批量插入
4. 对更新子集构造批量更新语句，避免逐个执行

示例代码（MySQL）：

import pandas as pd
import mysql.connector

api_data = [{"item_id": 1, "name": "test_update"}, {"item_id": 3, "name": "test_insert"}]
df = pd.DataFrame(api_data)

# 数据库连接
conn = mysql.connector.connect(host='localhost', database='your_db', user='your_user', password='your_pass')
cursor = conn.cursor()

# 批量查询已存在的item_id
item_ids = tuple(df['item_id'].tolist())
cursor.execute("SELECT item_id FROM your_table WHERE item_id IN %s", (item_ids,))
existing_ids = {row[0] for row in cursor.fetchall()}

# 拆分数据
insert_df = df[~df['item_id'].isin(existing_ids)]
update_df = df[df['item_id'].isin(existing_ids)]

# 批量插入新数据
if not insert_df.empty:
    insert_df.to_sql(
        name='your_table',
        con=conn,
        if_exists='append',
        index=False,
        method='multi',
        chunksize=500
    )

# 批量更新已有数据
if not update_df.empty:
    # 构造CASE WHEN批量更新语句
    update_query = """
        UPDATE your_table 
        SET name = CASE item_id 
    """
    cases = []
    params = []
    for _, row in update_df.iterrows():
        cases.append("WHEN %s THEN %s")
        params.extend([row['item_id'], row['name']])
    update_query += " ".join(cases) + " END WHERE item_id IN %s"
    params.append(tuple(update_df['item_id'].tolist()))
    
    cursor.execute(update_query, params)

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

方案3：临时表+批量同步 #

如果增量数据量稍大，用临时表先批量导入API数据，再通过数据库内部的同步操作完成UPSERT，性能会更优：

1. 创建与目标表结构一致的临时表
2. 把API数据批量插入临时表
3. 执行数据库级别的UPSERT，从临时表同步到目标表

示例代码（PostgreSQL）：

import pandas as pd
import psycopg2

api_data = [{"item_id": 1, "name": "test_update"}, {"item_id": 3, "name": "test_insert"}]
df = pd.DataFrame(api_data)

conn = psycopg2.connect("dbname=your_db user=your_user password=your_pass")
cur = conn.cursor()

# 创建临时表（会话结束自动销毁）
cur.execute("""
    CREATE TEMP TABLE temp_your_table (
        item_id INT PRIMARY KEY,
        name VARCHAR(255)
    ) ON COMMIT DROP;
""")

# 批量插入临时表
df.to_sql(
    name='temp_your_table',
    con=conn,
    if_exists='append',
    index=False,
    method='multi',
    chunksize=500
)

# 从临时表同步到目标表，完成UPSERT
cur.execute("""
    INSERT INTO your_table (item_id, name)
    SELECT item_id, name FROM temp_your_table
    ON CONFLICT (item_id) DO UPDATE SET
        name = EXCLUDED.name;
""")

conn.commit()
cur.close()
conn.close()

选择建议 #

• 优先用方案1：原生UPSERT+批量执行，代码简洁，网络开销最少，性能最优
• 若数据库不支持原生UPSERT，选方案2：拆分批次处理，避免逐个执行的低效
• 增量数据量较大时，选方案3：临时表的批量导入+数据库内部同步，IO效率更高

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