如何通过Python高效批量更新/插入Cloud SQL MySQL数据？

一、配置权限以使用LOAD DATA #

1. 启用LOCAL INFILE并授予FILE权限 #

如果要从本地CSV导入：

• 先在Cloud Console的Cloud SQL实例详情页，找到「数据库标志」，添加local_infile=ON，重启实例生效。
• 用超级用户（如root）给你的数据库用户授予全局FILE权限：

GRANT FILE ON *.* TO 'your_db_user'@'%';
FLUSH PRIVILEGES;

注意：Cloud SQL的FILE权限是全局的，不能限定到单个数据库，只有超级用户能操作。

2. 用GCS桶导入（推荐，无需额外用户权限） #

把CSV上传到GCS桶后，给Cloud SQL实例的服务账号添加Storage Object Viewer角色（在实例详情页的「服务账号」栏找到账号，去GCS桶的权限设置里添加），然后直接执行：

LOAD DATA INFILE 'gs://your-bucket/path/to/data.csv'
INTO TABLE your_table
FIELDS TERMINATED BY ',' ENCLOSED BY '"'
LINES TERMINATED BY '\n'
IGNORE 1 LINES
(id, val)
ON DUPLICATE KEY UPDATE val = VALUES(val);

二、高效批量Upsert的替代方案 #

1. SQLAlchemy批量执行INSERT...ON DUPLICATE KEY #

别循环执行单条语句，直接构造批量SQL：

from sqlalchemy import text

# data是包含(id, val)的列表，比如[(1, 'v1'), (2, 'v2'), ...]
batch_size = 10000  # 可根据实例性能调整，5000-20000之间测试最优值
for i in range(0, len(data), batch_size):
    batch = data[i:i+batch_size]
    placeholders = ', '.join(['(%s, %s)'] * len(batch))
    sql = f"""
        INSERT INTO your_table (id, val)
        VALUES {placeholders}
        ON DUPLICATE KEY UPDATE val = VALUES(val)
    """
    # 用原生SQL执行，避开ORM的额外开销
    session.execute(text(sql), [item for sublist in batch for item in sublist])
session.commit()

这种方式把多次网络请求合并成批量，40万行应该能压缩到几分钟内完成。

2. 临时表中转合并 #

适合超大数据量的场景：

1. 创建临时表，去掉不必要的索引加快导入：

CREATE TEMPORARY TABLE temp_table LIKE your_table;
ALTER TABLE temp_table DROP INDEX non_unique_index_name;  # 删掉非唯一索引

2. 用LOAD DATA或批量INSERT把所有数据导入临时表
3. 一次性合并到目标表：

INSERT INTO your_table (id, val)
SELECT id, val FROM temp_table
ON DUPLICATE KEY UPDATE val = VALUES(val);

4. 临时表会在会话结束后自动销毁，也可以手动删除：

DROP TEMPORARY TABLE temp_table;

3. 跳过无变化的行 #

如果你的计算流程能提前对比新旧值，只收集需要插入/更新的行：

# 假设已从数据库读取现有id和val的映射：existing = {id: current_val}
to_upsert = []
for item in data:
    item_id, new_val = item
    if item_id not in existing or existing[item_id] != new_val:
        to_upsert.append(item)
# 对to_upsert执行批量Upsert

这个方法适合大部分行val没有变化的场景，能减少数据库处理的总行数。

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