将一百万条数据高效写入数据库是数据工程中的常见挑战。关键在于**减少网络往返、利用批量操作、合理管理事务、选择适当的工具与API**。根据数据库类型（如MySQL、PostgreSQL）和Python库（如`pymysql`、`psycopg2`、`SQLAlchemy`、`pandas`），有多种优化方案。以下对比了主流方法的核心差异，并提供了具体的代码实现。 #### **核心方法对比与选择** 下表从原理、性能、适用场景和关键注意事项对几种主流方法进行了对比[ref_1][ref_3][ref_4][ref_6]。 | 方法 | 核心原理 | 性能评价 | 适用场景 | 关键注意事项 | | :--- | :--- | :--- | :--- | :--- | | **逐条插入 (`cursor.execute()`)** | 每条数据生成一条独立的`INSERT` SQL语句，并单独提交（或在一个事务内）[ref_6]。 | **极差**。产生海量网络请求和SQL解析开销，效率最低。 | 仅用于理解原理或极少量数据（<1000条）。 | 绝对禁止用于百万级数据写入。 | | **标准批量插入 (`cursor.executemany()`)** | 将多条`INSERT`语句的占位符和参数一次性发送给数据库执行[ref_3][ref_6]。 | **中等**。相比逐条有显著提升，但仍是多次解析的模拟，数据库优化有限。 | 万级数据量，或数据库驱动原生支持批量协议。 | 注意单次提交的数据量不宜过大，避免内存或网络问题。 | | **高性能扩展 (`psycopg2.extras.execute_values()`)** | 将多条数据合并为一条带有多个`VALUES`子句的扩展SQL语句[ref_3][ref_4]。 | **高**。显著减少SQL解析和网络开销，是PostgreSQL批量写入的推荐方法之一[ref_4]。 | PostgreSQL数据库，十万至百万级数据。 | 仅适用于PostgreSQL的`psycopg2`库。需关注SQL语句长度限制。 | | **COPY命令 (`cursor.copy_from()` / `copy_to`)** | 使用数据库原生的二进制或文本COPY协议进行数据流式导入[ref_2]。 | **极高**。绕过SQL解析层，是关系型数据库最高效的导入方式之一。 | PostgreSQL、MySQL（`LOAD DATA INFILE`）等支持COPY的数据库，百万级以上数据。 | 对数据格式要求严格，通常需要先将数据写入内存缓冲区（如`StringIO`）。 | | **Pandas批量写入 (`df.to_sql()`)** | `pandas`通过`SQLAlchemy`引擎，内部可能使用`multi`模式（多值插入）或逐条插入[ref_1][ref_3]。 | **中等至高（取决于参数）**。使用方便，但性能取决于底层引擎和`method`参数。 | 数据已存在于`DataFrame`中，追求开发效率，数据量在十万至百万级。 | 默认方式（`method=None`）可能很慢。推荐设置`if_exists='append'`和`chunksize`，并可自定义`method`。 | #### **具体实现与代码示例** **1. 使用 `psycopg2` 与 `execute_values` 写入 PostgreSQL** 这是PostgreSQL中平衡性能与易用性的首选方法[ref_3][ref_4]。 ```python import psycopg2 from psycopg2 import extras # 1. 建立数据库连接 conn = psycopg2.connect( host="your_host", database="your_db", user="your_user", password="your_password" ) cur = conn.cursor() # 2. 模拟生成一百万条数据 (例如：ID和随机数) data = [(i, f"user_{i}", i * 10) for i in range(1, 1000001)] # 3. 使用 execute_values 进行批量插入 # `page_size` 参数控制每批插入的数据量，建议在1000-10000之间调整 insert_query = "INSERT INTO your_table (id, name, value) VALUES %s" extras.execute_values(cur, insert_query, data, page_size=5000) # 4. 提交事务并关闭连接 conn.commit() cur.close() conn.close() ``` **关键说明**：`execute_values` 函数内部会将数据分批构建成 `INSERT INTO ... VALUES (1,'a'), (2,'b'), ...` 的SQL形式，极大提升了效率[ref_4]。 **2. 使用 `psycopg2` 与 `copy_from` 写入 PostgreSQL（最高性能）** 对于百万级数据，这是最快的本地方法，利用了PostgreSQL的原生COPY协议[ref_2]。 ```python import psycopg2 from io import StringIO # 建立连接 conn = psycopg2.connect("...") cur = conn.cursor() # 生成内存中的文件对象，并将数据写入其中，格式需符合COPY要求（如制表符分隔） output = StringIO() for item in data: # data为包含百万条记录的列表 # 每行数据以制表符分隔，并以换行符结尾 output.write(f"{item[0]}\t{item[1]}\t{item[2]}\n") output.seek(0) # 将指针移回文件开头 # 使用 copy_from 从内存文件对象导入数据 cur.copy_from(output, 'your_table', columns=('id', 'name', 'value')) conn.commit() cur.close() conn.close() ``` **3. 使用 `pymysql` 进行标准批量写入（MySQL）** 对于MySQL，可使用`cursor.executemany`方法[ref_1][ref_6]。 ```python import pymysql # 建立连接 conn = pymysql.connect(host='localhost', user='root', password='pass', database='test') cursor = conn.cursor() # 准备数据 data = [(i, f"user_{i}", i * 10) for i in range(1, 1000001)] # 定义SQL，使用参数化查询防止SQL注入 sql = "INSERT INTO your_table (id, name, value) VALUES (%s, %s, %s)" # 分批执行 executemany，避免单次数据量过大 batch_size = 5000 for i in range(0, len(data), batch_size): batch = data[i:i+batch_size] cursor.executemany(sql, batch) conn.commit() # 每批提交一次 cursor.close() conn.close() ``` **4. 使用 `pandas.DataFrame.to_sql` 通用方法** 如果你的数据已经在`pandas DataFrame`中，此方法最为便捷。通过调整参数可优化性能[ref_1][ref_3]。 ```python import pandas as pd from sqlalchemy import create_engine # 1. 创建示例DataFrame (模拟一百万条数据) df = pd.DataFrame({ 'id': range(1, 1000001), 'name': [f'user_{i}' for i in range(1, 1000001)], 'value': range(10, 10000010, 10) }) # 2. 创建SQLAlchemy引擎 engine = create_engine('postgresql://user:pass@host:port/dbname') # MySQL: mysql+pymysql://... # 3. 使用to_sql写入，关键优化参数： # if_exists='append': 追加到现有表 # index=False: 不写入DataFrame索引 # chunksize: 分块写入，每块5000条 # method='multi': 指示数据库驱动使用多值插入（部分数据库支持，可显著提升速度） df.to_sql('your_table', engine, if_exists='append', index=False, chunksize=5000, method='multi') ``` **注意**：`method='multi'` 并非所有数据库驱动都支持，对于PostgreSQL的`psycopg2`，它相当于生成多值`VALUES`子句的语句。对于不支持的情况，可考虑自定义`method`，例如使用`psycopg2.extras.execute_values`[ref_3]。 #### **性能优化最佳实践总结** 1. **分批提交**：无论使用哪种方法，都应避免一次性提交全部一百万条数据。将数据分成大小为 **1000 到 10000** 的批次进行提交，可以平衡内存使用、网络传输和事务锁的持有时间[ref_1][ref_6]。 2. **关闭自动提交**：在批量操作前，确保连接设置为**手动提交模式**（`autocommit=False`），在一个批次成功插入后再执行`conn.commit()`。这能确保数据一致性，并在出错时便于回滚[ref_1]。 3. **利用数据库原生协议**：在追求极致性能时，应优先考虑数据库提供的**原生批量导入工具**，如PostgreSQL的`COPY`[ref_2]、MySQL的`LOAD DATA INFILE`或`INSERT ... VALUES (),(),...`。对应的Python驱动（如`psycopg2.copy_from`）能直接调用这些功能。 4. **索引与约束管理**：如果目标表上有大量索引、外键约束或触发器，会在插入时带来额外开销。对于一次性历史数据导入，可考虑**先删除非关键索引和约束，导入完成后再重建**，能大幅提升写入速度。 5. **连接池与并发**：对于超大规模数据，单线程写入可能达到瓶颈。可以考虑使用**连接池**配合**多进程/多线程**（需注意数据库连接线程安全性），将数据分区后并行写入不同的数据库连接。但这增加了复杂度，需谨慎处理数据分区和事务隔离问题。