本文目录导读：
我来介绍几种批量插入数据的方法,根据不同的数据库和场景。
MySQL 批量插入
SQL 语句方式
-- 一条语句插入多条记录
INSERT INTO users (name, age, email) VALUES
('张三', 25, 'zhangsan@example.com'),
('李四', 30, 'lisi@example.com'),
('王五', 28, 'wangwu@example.com'),
('赵六', 35, 'zhaoliu@example.com');
Python 脚本示例
import mysql.connector
from mysql.connector import Error
def batch_insert_mysql(data_list, batch_size=1000):
try:
connection = mysql.connector.connect(
host='localhost',
database='test_db',
user='root',
password='password'
)
cursor = connection.cursor()
# 分批插入
for i in range(0, len(data_list), batch_size):
batch = data_list[i:i+batch_size]
sql = "INSERT INTO users (name, age, email) VALUES (%s, %s, %s)"
cursor.executemany(sql, batch)
connection.commit()
print(f"已插入 {i + len(batch)} 条记录")
except Error as e:
print(f"错误: {e}")
connection.rollback()
finally:
if connection.is_connected():
cursor.close()
connection.close()
# 准备数据
data = [('用户%d' % i, 20 + i%50, 'user%d@example.com' % i)
for i in range(10000)]
batch_insert_mysql(data)
PostgreSQL 批量插入
使用 COPY 命令（最快方式）
import psycopg2
from io import StringIO
def batch_insert_postgresql(data_list):
conn = psycopg2.connect(
host='localhost',
database='test_db',
user='postgres',
password='password'
)
cursor = conn.cursor()
# 使用 StringIO 模拟文件
buffer = StringIO()
for name, age, email in data_list:
buffer.write(f"{name}\t{age}\t{email}\n")
buffer.seek(0)
cursor.copy_from(buffer, 'users',
columns=('name', 'age', 'email'),
sep='\t')
conn.commit()
cursor.close()
conn.close()
SQLite 批量插入
import sqlite3
def batch_insert_sqlite(data_list, batch_size=500):
conn = sqlite3.connect('test.db')
cursor = conn.cursor()
# 开启事务
cursor.execute('BEGIN TRANSACTION')
try:
for i in range(0, len(data_list), batch_size):
batch = data_list[i:i+batch_size]
sql = "INSERT INTO users (name, age, email) VALUES (?, ?, ?)"
cursor.executemany(sql, batch)
conn.commit()
print(f"成功插入 {len(data_list)} 条记录")
except Exception as e:
conn.rollback()
print(f"错误: {e}")
finally:
conn.close()
使用 SQL 脚本生成
生成批量 INSERT 脚本
def generate_insert_script(data_list, table_name='users', batch_size=1000):
scripts = []
for i in range(0, len(data_list), batch_size):
batch = data_list[i:i+batch_size]
values = []
for name, age, email in batch:
# 转义单引号
name = name.replace("'", "''")
email = email.replace("'", "''")
values.append(f"('{name}', {age}, '{email}')")
sql = f"INSERT INTO {table_name} (name, age, email) VALUES\n"
sql += ",\n".join(values) + ";"
scripts.append(sql)
return scripts
# 生成10000条数据的脚本
data = [('用户%d' % i, 20 + i%50, 'user%d@example.com' % i)
for i in range(10000)]
scripts = generate_insert_script(data)
# 保存到文件
with open('batch_insert.sql', 'w') as f:
for script in scripts:
f.write(script + '\n\n')
Shell 脚本批量插入
#!/bin/bash
# 批量插入 MySQL
batch_insert_mysql() {
local batch_size=1000
local total=10000
# 生成数据并插入
for ((i=0; i<total; i+=batch_size)); do
sql="INSERT INTO users (name, age, email) VALUES "
values=()
for ((j=i; j<i+batch_size && j<total; j++)); do
values+=("('用户$j', $((20 + j % 50)), 'user$j@example.com')")
done
sql+=$(IFS=,; echo "${values[*]}")
mysql -u root -p'password' test_db -e "$sql"
echo "已插入 $((i + batch_size)) 条记录"
done
}
batch_insert_mysql
性能优化建议
开启批量模式
-- MySQL SET GLOBAL bulk_insert_buffer_size = 256 * 1024 * 1024; -- 关闭索引更新（插完再重建） ALTER TABLE users DISABLE KEYS; -- 执行批量插入 ALTER TABLE users ENABLE KEYS;
事务管理
# 每批次提交一次事务
def optimized_batch_insert(data_list, batch_size=1000):
conn = get_connection()
cursor = conn.cursor()
for i in range(0, len(data_list), batch_size):
batch = data_list[i:i+batch_size]
try:
conn.begin()
for item in batch:
cursor.execute("INSERT INTO ...", item)
conn.commit()
except Exception as e:
conn.rollback()
print(f"批次 {i//batch_size} 插入失败: {e}")
使用 ORM 批量插入
SQLAlchemy 示例
from sqlalchemy import create_engine, Table, MetaData, Column, Integer, String
from sqlalchemy.orm import Session
def orm_batch_insert(data_list):
engine = create_engine('mysql+pymysql://root:password@localhost/test_db')
metadata = MetaData()
users_table = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(50)),
Column('age', Integer),
Column('email', String(100))
)
with Session(engine) as session:
session.execute(
users_table.insert(),
data_list
)
session.commit()
注意事项
- 批量大小：一般1000-5000条/批次
- 内存管理：避免一次加载太多数据
- 事务控制：每批次提交一次
- 索引策略：批量插入后重建索引
- 错误处理：实现完善的异常处理
选择哪种方式取决于你的具体场景、数据库类型和性能要求。
