本文旨在提供在Python和Django环境下，向PostgreSQL数据库高效导入海量数据的策略与实践。针对传统批处理插入可能面临的性能瓶颈和连接中断问题，文章详细介绍了两种优化方案：利用数据库会话的预处理语句（Prepared Statements）提升重复插入效率，以及采用PostgreSQL原生的`COPY`命令实现极致的导入速度。同时，文章还将探讨相关最佳实践，包括事务管理、索引优化及与Django框架的集成考量。
在处理大规模数据导入PostgreSQL时，传统的逐行插入或简单的多值INSERT语句可能无法满足性能要求，甚至会导致OperationalError: server closed the connection unexpectedly之类的连接问题。本教程将深入探讨如何利用PostgreSQL的特性，结合Python和Django环境，实现高效、稳定的海量数据导入。
当前常见的批处理插入方法，如通过INSERT INTO ... VALUES (v1), (v2), ...的形式一次性插入多行数据，确实比单行插入效率更高。然而，当数据量达到百万级别或更高时，这种方法仍可能面临以下挑战：
预处理语句允许数据库对一个SQL模板进行一次性解析、规划和优化，然后可以多次执行，只需提供不同的参数。这对于重复执行相同结构但参数不同的插入操作非常有效。
虽然Django ORM通常不会直接暴露预处理语句的API，但通过直接操作psycopg2游标，我们可以实现这一优化。
工作原理：
Python/psycopg2 示例：
from django.db import connection
import psycopg2
def insert_with_prepared_statement(data_batches):
with connection.cursor() as cursor:
# 获取底层的psycopg2连接和游标
pg_conn = cursor.connection
pg_cursor = pg_conn.cursor()
try:
# 1. 准备语句
# 使用psycopg2的execute()方法执行PREPARE命令
# 声明一个名为'my_insert_stmt'的预处理语句
# 参数类型需要明确指定，例如TEXT, INT, BIGINT等
pg_cursor.execute("""
PREPARE my_insert_stmt (BIGINT, TEXT, BIGINT, NUMERIC) AS
INSERT INTO per_tr
ansaction_table (company_ref_id_id_column, rrn_column, transaction_type_ref_id_id_column, transactionamount_column)
VALUES ($1, $2, $3, $4)
ON CONFLICT (rrn_column) DO UPDATE SET company_ref_id_id_column = EXCLUDED.company_ref_id_id_column;
""")
for batch in data_batches:
# 开启一个事务块，确保批次内的操作原子性
pg_cursor.execute("BEGIN;")
for row_data in batch:
# 2. 执行语句
# 使用EXECUTE命令调用预处理语句，并传入参数
pg_cursor.execute("EXECUTE my_insert_stmt (%s, %s, %s, %s);", row_data)
pg_cursor.execute("COMMIT;")
print(f"Successfully inserted {len(data_batches) * len(data_batches[0])} rows using prepared statements.")
except psycopg2.Error as e:
pg_conn.rollback() # 发生错误时回滚
print(f"Database error: {e}")
finally:
# 3. 释放语句 (可选，会话结束时会自动释放)
pg_cursor.execute("DEALLOCATE my_insert_stmt;")
pg_cursor.close()
# 示例数据生成 (假设数据已包含计算结果)
# data_batches 应该是一个列表的列表，每个内部列表代表一个批次，每个批次包含多个元组，每个元组代表一行数据
# 例如：[[ (1, 'R1', 101, 100.50), (2, 'R2', 102, 200.75) ], ...]
# 假设 company_ref_id_id_column 为 BIGINT, rrn_column 为 TEXT, transaction_type_ref_id_id_column 为 BIGINT, transactionamount_column 为 NUMERIC
# (请根据实际表结构调整参数类型和顺序)
# example_data_batches = [
# [(1, 'R1', 101, 100.50), (2, 'R2', 102, 200.75)],
# [(3, 'R3', 103, 300.25), (4, 'R4', 104, 400.00)]
# ]
# insert_with_prepared_statement(example_data_batches)注意事项：
COPY命令是PostgreSQL提供的一种最高效的数据导入方式，它允许直接在服务器端进行数据传输，绕过了SQL解析器的大部分开销。它比任何INSERT语句都快，因为它是为批量加载而设计的。
COPY命令支持从文件导入 (COPY FROM filename) 或从标准输入导入 (COPY FROM STDIN)。对于Python应用程序，COPY FROM STDIN是最常用的方式，通过psycopg2的copy_from或copy_expert方法实现。
COPY命令的优势：
Python/psycopg2 COPY FROM STDIN 示例：
ansaction_table (company_ref_id_id_column, rrn_column, transaction_type_ref_id_id_column, transactionamount_column)
VALUES ($1, $2, $3, $4)
ON CONFLICT (rrn_column) DO UPDATE SET company_ref_id_id_column = EXCLUDED.company_ref_id_id_column;
""")
for batch in data_batches:
# 开启一个事务块，确保批次内的操作原子性
pg_cursor.execute("BEGIN;")
for row_data in batch:
# 2. 执行语句
# 使用EXECUTE命令调用预处理语句，并传入参数
pg_cursor.execute("EXECUTE my_insert_stmt (%s, %s, %s, %s);", row_data)
pg_cursor.execute("COMMIT;")
print(f"Successfully inserted {len(data_batches) * len(data_batches[0])} rows using prepared statements.")
except psycopg2.Error as e:
pg_conn.rollback() # 发生错误时回滚
print(f"Database error: {e}")
finally:
# 3. 释放语句 (可选，会话结束时会自动释放)
pg_cursor.execute("DEALLOCATE my_insert_stmt;")
pg_cursor.close()
# 示例数据生成 (假设数据已包含计算结果)
# data_batches 应该是一个列表的列表，每个内部列表代表一个批次，每个批次包含多个元组，每个元组代表一行数据
# 例如：[[ (1, 'R1', 101, 100.50), (2, 'R2', 102, 200.75) ], ...]
# 假设 company_ref_id_id_column 为 BIGINT, rrn_column 为 TEXT, transaction_type_ref_id_id_column 为 BIGINT, transactionamount_column 为 NUMERIC
# (请根据实际表结构调整参数类型和顺序)
# example_data_batches = [
# [(1, 'R1', 101, 100.50), (2, 'R2', 102, 200.75)],
# [(3, 'R3', 103, 300.25), (4, 'R4', 104, 400.00)]
# ]
# insert_with_prepared_statement(example_data_batches)import io
from django.db import connection
import psycopg2
def insert_with_copy_command(data_generator):
with connection.cursor() as cursor:
pg_conn = cursor.connection
pg_cursor = pg_conn.cursor()
try:
# 使用StringIO模拟文件，将数据格式化为CSV或TSV
# 确保数据的顺序与目标表的列顺序一致
# 如果有ON CONFLICT需求，需要使用COPY FROM PROGRAM 或 copy_expert 结合临时表
# 或者先COPY到临时表，再从临时表进行UPSERT
# 这里先展示最简单的COPY，不带ON CONFLICT
output = io.StringIO()
for row_data in data_generator:
# 假设数据是 (company_ref_id, rrn, transaction_type_ref_id, transaction_amount)
# 并且 rrn_column 是文本类型，其他是数字
# 格式化为CSV格式，逗号分隔，文本字段加引号
output.write(f"{row_data[0]},\"{row_data[1]}\",{row_data[2]},{row_data[3]}\n")
output.seek(0) # 将文件指针移到开头
# 执行COPY命令
# 注意：如果表中有ON CONFLICT，COPY INTO TABLE 无法直接处理。
# 通常的做法是COPY到临时表，然后从临时表进行UPSERT。
# 或者使用COPY FROM PROGRAM并结合SQL语句，但更复杂。
# 对于有ON CONFLICT的场景，推荐先COPY到临时表，再进行MERGE/UPSERT。
# 简单的COPY示例 (无ON CONFLICT)
table_name = "per_transaction_table" # 替换为你的表名
columns = "(company_ref_id_id_column, rrn_column, transaction_type_ref_id_id_column, transactionamount_column)"
# 使用copy_expert来处理更复杂的COPY选项，例如CSV格式
pg_cursor.copy_expert(
f"COPY {table_name} {columns} FROM STDIN WITH (FORMAT CSV, DELIMITER ',', QUOTE '\"');",
output
)
pg_conn.commit() # COPY操作通常需要在一个事务中
print(f"Successfully inserted data using COPY command.")
except psycopg2.Error as e:
pg_conn.rollback()
print(f"Database error during COPY: {e}")
finally:
pg_cursor.close()
# 示例数据生成器 (假设数据已包含计算结果)
# def generate_large_data(num_rows):
# for i in range(num_rows):
# yield (i + 1, f'R{i+1:07d}', (i % 10) + 100, (i + 1) * 10.50)
#
# insert_with_copy_command(generate_large_data(1000000))处理ON CONFLICT与COPY：
COPY命令本身不直接支持ON CONFLICT。如果需要处理冲突（UPSERT），通常有以下策略：
COPY到临时表，然后UPSERT：
def insert_with_copy_and_upsert(data_generator, target_table_name, conflict_column, columns_to_insert):
with connection.cursor() as cursor:
pg_conn = cursor.connection
pg_cursor = pg_conn.cursor()
temp_table_name = f"temp_{target_table_name}_{pg_conn.pid}" # 使用进程ID避免冲突
try:
# 1. 创建临时表 (结构与目标表一致)
pg_cursor.execute(f"""
CREATE TEMPORARY TABLE {temp_table_name} (LIKE {target_table_name} INCLUDING DEFAULTS);
""")
# 2. 准备数据并COPY到临时表
output = io.StringIO()
for row_data in data_generator:
# 确保数据格式与temp_table_name的列匹配
output.write(",".join(map(str, row_data)) + "\n") # 简单示例，实际需根据数据类型做CSV/TSV格式化
output.seek(0)
pg_cursor.copy_expert(
f"COPY {temp_table_name} ({','.join(columns_to_insert)}) FROM STDIN WITH (FORMAT CSV, DELIMITER ',');",
output
)
# 3. 从临时表进行UPSERT到目标表
update_set_clause = ", ".join([f"{col} = EXCLUDED.{col}" for col in columns_to_insert if col != conflict_column])
pg_cursor.execute(f"""
INSERT INTO {target_table_name} ({','.join(columns_to_insert)})
SELECT {','.join(columns_to_insert)} FROM {temp_table_name}
ON CONFLICT ({conflict_column}) DO UPDATE SET {update_set_clause};
""")
pg_conn.commit()
print(f"Successfully inserted/updated data using COPY to temp table and UPSERT.")
except psycopg2.Error as e:
pg_conn.rollback()
print(f"Database error during COPY+UPSERT: {e}")
finally:
# 4. 删除临时表
pg_cursor.execute(f"DROP TABLE IF EXISTS {temp_table_name};")
pg_cursor.close()
# 示例调用
# columns_to_insert = ['company_ref_id_id_column', 'rrn_column', 'transaction_type_ref_id_id_column', 'transactionamount_column']
# conflict_col = 'rrn_column'
# insert_with_copy_and_upsert(generate_large_data(1000000), 'per_transaction_table', conflict_col, columns_to_insert)除了选择合适的导入方法，还有一些通用的最佳实践可以进一步提升性能：
如果数据是临时的、可以随时重建的，并且不需要WAL日志记录（即不需要崩溃恢复或流复制），可以考虑使用UNLOGGED TABLE。
优势： UNLOGGED表不会写入WAL日志，这使得数据写入速度极快。
劣势： 数据库崩溃时，非日志表内容会丢失。它们也不能用于流复制或时间点恢复。
使用场景： 作为临时数据暂存区，导入后进行处理或聚合，然后将结果写入永久表。
CREATE UNLOGGED TABLE my_temp_data (
id BIGINT,
name TEXT
);对于COPY到UNLOGGED表，可以使用WITH (FREEZE)选项进一步优化，但这通常在COPY到空表时效果显著，且需谨慎使用，因为它会标记行已冻结，跳过VACUUM检查。
在Python和Django环境中向PostgreSQL导入海量数据时，选择合适的策略至关重要。
通过综合运用这些技术，可以有效地解决海量数据导入PostgreSQL所面临的性能和稳定性挑战。始终建议在实际生产环境前，在测试环境中进行充分的性能测试和调优。