如何优化Python跨两个MySQL实例数据传输的执行速度

现有代码核心性能瓶颈 #

• 全量数据一次性加载到内存：pd.read_sql未配置分块读取参数，会将所有查询结果拉取到本地内存生成完整DataFrame，数据量较大时不仅内存占用高，单线程IO等待周期长，还会额外触发pandas全量数据类型推断的冗余开销
• 写入逻辑效率极差：pandas.to_sql默认逐行生成单条INSERT语句，即使配置了chunksize=1000，本质仍是循环执行单条插入，完全没有利用MySQL多值批量插入能力，每条语句都存在网络往返、事务刷盘的固定开销
• 导入了多进程模块但完全未使用，读写全流程单线程串行执行，硬件资源利用率极低
• 数据库连接未做针对性优化：SQLAlchemy连接串未配置批量操作相关参数，存在不必要的隐式转换、事务开销

按收益从高到低的可落地方案 #

1. 替换默认写入逻辑，启用MySQL批量插入 #

这是改动最小、收益最高的优化点，通常能带来10~100倍的写入性能提升：

• 创建目标数据库引擎时开启fast_executemany=True参数，驱动会自动将多条插入语句合并为单条多值INSERT格式（即INSERT INTO t (c1,c2) VALUES (1,2),(3,4)...），大幅减少网络交互和SQL解析开销
• 给to_sql传入method='multi'参数，配合合理的批次大小，避免单条SQL过长触发MySQLmax_allowed_packet限制

2. 改全量加载为流式读写，降低内存占用同时压缩等待时间 #

不要等所有数据全部抽取完成再执行写入，配置pd.read_sql的chunksize参数（建议设为5000~20000，根据网络带宽调整），读一批、处理一批、写一批，将全流程从串行改为流水线模式，内存占用从O(全量数据)降到O(单批次数据)，同时减少抽数阶段的空等时间。
固定值字段比如load_id直接在源端SELECT语句中拼接，不要拉取到本地后再给DataFrame加列，省掉pandas列操作的开销。

3. 优化连接与事务配置，减少冗余开销 #

• 连接串显式指定和数据库一致的字符集（比如charset=utf8mb4），避免驱动做隐式字符集转换
• 关闭目标端连接的自动提交，每写入5~10个批次统一提交一次事务，避免单批次一次提交的刷盘开销，同时避免长事务锁表
• 源端查询只返回需要的列，过滤条件全部下推到源端SQL执行，绝对不要用SELECT *拉全量字段到本地再用pandas过滤

4. 百万级以上数据启用并行分片同步 #

数据量超过百万行时，可以用导入的multiprocessing模块做并行同步：选择源表的均匀分片键（比如自增主键、时间字段），先查询分片键的最大、最小值，将全量查询拆分为N个范围不重叠的子查询，每个进程负责一个分片的读写，进程数设置为CPU核心数和数据库最大连接允许数的最小值（通常4~8即可，不要开过多进程打满数据库负载）。

5. 千万级以上数据直接用MySQL原生导入能力 #

单表数据量过千万时，直接放弃pandas层的读写逻辑，在Python中调用MySQL原生能力：先执行SELECT ... INTO OUTFILE将查询结果导出为CSV临时文件，再执行LOAD DATA LOCAL INFILE将CSV导入目标实例，该方式比pandas批量插入还要快5~10倍，是MySQL官方推荐的大数据量导入方案。

配套优化手段 #

• 同步前临时关闭目标表的二级索引、外键约束，等数据全部写入完成后再重建，避免InnoDB每插入一行就更新所有索引的开销
• 同步前可以临时调大目标实例的innodb_flush_log_at_trx_commit和sync_binlog参数（数据同步完成后改回原值），减少事务刷盘频率
• 单批插入的行数控制在1000~5000区间，避免单条SQL长度超过max_allowed_packet导致报错

优化后参考实现 #

import pandas as pd
from sqlalchemy import create_engine

# 同步配置
READ_CHUNKSIZE = 10000
WRITE_BATCH_SIZE = 2000
COMMIT_INTERVAL = 5
LOAD_ID = "2"
# 固定字段直接在SQL中拼接，过滤条件全部下推到源端
EXTRACT_SQL = f"""
    SELECT col1, col2, col3, '{LOAD_ID}' AS load_id 
    FROM source_table 
    WHERE 你的业务过滤条件
"""

if __name__ == "__main__":
    try:
        # 初始化优化后的数据库连接
        engine_source = create_engine(
            "mysql+mysqlconnector://账号:密码@源库IP:端口/源库名?charset=utf8mb4",
            pool_pre_ping=True
        )
        engine_dest = create_engine(
            "mysql+mysqlconnector://账号:密码@目标库IP:端口/目标库名?charset=utf8mb4&allow_local_infile=1",
            fast_executemany=True,
            pool_pre_ping=True
        )

        with engine_source.connect() as conn_source, engine_dest.connect() as conn_dest:
            conn_dest.execution_options(autocommit=False)
            synced_rows = 0
            # 流式读取+批量写入
            for chunk in pd.read_sql(EXTRACT_SQL, conn_source, chunksize=READ_CHUNKSIZE):
                chunk.to_sql(
                    schema="目标库名",
                    name="目标表名",
                    con=conn_dest,
                    if_exists="append",
                    index=False,
                    method="multi",
                    chunksize=WRITE_BATCH_SIZE
                )
                synced_rows += len(chunk)
                # 定期提交事务，避免长事务
                if synced_rows % (READ_CHUNKSIZE * COMMIT_INTERVAL) == 0:
                    conn_dest.commit()
                    print(f"已同步数据量：{synced_rows}行")
            # 提交最后一批剩余数据
            conn_dest.commit()
        
        engine_source.dispose()
        engine_dest.dispose()
        print(f"数据同步完成，总同步量：{synced_rows}行")

    except Exception as e:
        print(f"同步任务异常：{str(e)}")
        raise

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