脚本如何批量插入数据

本文目录导读：

1. MySQL 批量插入
2. PostgreSQL 批量插入
3. SQLite 批量插入
4. 使用 SQL 脚本生成
5. Shell 脚本批量插入
6. 性能优化建议
7. 使用 ORM 批量插入
8. 注意事项

我来介绍几种批量插入数据的方法,根据不同的数据库和场景。

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()

注意事项

1. 批量大小：一般1000-5000条/批次
2. 内存管理：避免一次加载太多数据
3. 事务控制：每批次提交一次
4. 索引策略：批量插入后重建索引
5. 错误处理：实现完善的异常处理

选择哪种方式取决于你的具体场景、数据库类型和性能要求。