如何加速Pandas Dataframe写入MySQL数据库的INSERT查询

1. 更换高性能数据库驱动 #

mysqlconnector的性能普遍低于mysqlclient（对应SQLAlchemy的mysql+mysqldb驱动），更换后能显著提升批量写入速度。

安装依赖：

pip install mysqlclient

修改引擎初始化代码：

from sqlalchemy import create_engine

# 替换为mysql+mysqldb驱动
engine = create_engine('mysql+mysqldb://xxxx:xxxxx@localhost:3306/db_qa', pool_pre_ping=True)

2. 启用SQLAlchemy批量插入优化 #

方式A：使用fast_executemany参数 #

配合mysqlclient驱动启用批量插入优化，同时调整连接池参数减少连接开销：

engine = create_engine(
    'mysql+mysqldb://xxxx:xxxxx@localhost:3306/db_qa',
    fast_executemany=True,
    pool_size=10,
    max_overflow=20
)

df.to_sql('my_table', con=engine, if_exists='replace', index=False, chunksize=1000)

方式B：手动控制事务 #

将整个写入过程包裹在单个事务中，减少频繁提交的开销：

with engine.begin() as conn:
    df.to_sql('my_table', con=conn, if_exists='replace', index=False, chunksize=1000, method='multi')

3. 使用MySQL原生LOAD DATA INFILE导入 #

这是MySQL最快的批量导入方式，性能远高于常规INSERT批量写入。步骤为先将DataFrame导出为内存CSV，再执行原生导入命令：

import io
import csv
from sqlalchemy import text

# 导出DataFrame到内存CSV
csv_buffer = io.StringIO()
df.to_csv(csv_buffer, sep='\t', na_rep='\\N', index=False, quoting=csv.QUOTE_NONE)
csv_buffer.seek(0)

with engine.connect() as conn:
    # 先创建空表（复用DataFrame的结构）
    df.head(0).to_sql('my_table', con=conn, if_exists='replace', index=False)
    # 执行LOAD DATA命令
    conn.execute(
        text("""
            LOAD DATA LOCAL INFILE '/dev/stdin' INTO TABLE my_table
            FIELDS TERMINATED BY '\t'
            OPTIONALLY ENCLOSED BY '"'
            LINES TERMINATED BY '\n'
            IGNORE 1 LINES;
        """),
        execution_options={"raw": True}
    )
    conn.commit()

注意：需要确保MySQL实例开启local_infile=1参数，并且引擎初始化时添加connect_args={'local_infile': True}。

4. 优化Cloud SQL Proxy连接 #

• 使用Unix Socket代替TCP：若VM与Cloud SQL实例同区域，用Unix Socket连接可大幅降低网络延迟：

  # 替换为你的Cloud SQL实例连接名对应的Socket路径
  engine = create_engine('mysql+mysqldb://xxxx:xxxxx@/db_qa?unix_socket=/cloudsql/your-project:region:instance-name')
  

• 调整连接池参数：增加连接池大小，避免频繁创建销毁连接：

  engine = create_engine(
      'mysql+mysqldb://xxxx:xxxxx@localhost:3306/db_qa',
      pool_size=15,
      max_overflow=30,
      pool_recycle=3600
  )
  

5. 临时优化MySQL实例参数 #

针对导入场景临时调整InnoDB参数，导入完成后建议改回原配置：

-- 临时关闭日志实时刷盘
SET GLOBAL innodb_flush_log_at_trx_commit = 2;
-- 临时禁用唯一键检查
SET UNIQUE_CHECKS = 0;
-- 临时禁用外键检查
SET FOREIGN_KEY_CHECKS = 0;

导入完成后恢复：

SET GLOBAL innodb_flush_log_at_trx_commit = 1;
SET UNIQUE_CHECKS = 1;
SET FOREIGN_KEY_CHECKS = 1;

6. 使用第三方优化库 #

d6tstack专门优化Pandas数据导入数据库的性能，可直接调用封装好的高效导入方法：

pip install d6tstack

import d6tstack

d6tstack.utils.pd_to_mysql(df, 'mysql+mysqldb://xxxx:xxxxx@localhost:3306/db_qa', 'my_table', if_exists='replace')

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