在Python中批量处理大数据写入数据库是一种常见的做法，目的是为了提升数据写入效率和减少系统资源消耗。以下是几个关键步骤和策略：
1. 准备数据
使用Pandas或其他数据处理库加载或生成数据，将其转换为DataFrame或其他易于处理的格式。
对数据进行预处理，包括清理、转换和验证，确保数据满足数据库表结构的要求。
2. 连接数据库
使用适当的数据库驱动器建立连接，例如使用sqlite3连接SQLite数据库，psycopg2连接PostgreSQL数据库，pyodbc或pymysql连接MySQL数据库等。
3. 批量插入
使用pandas的to_sql方法
对于小到中等规模的数据，pandas的to_sql方法可以直接将DataFrame批量写入数据库，它内部通常会执行批量插入操作：
Python 1import pandas as pd 2from sqlalchemy import create_engine 3 4# 创建数据库引擎 5engine = create_engine('postgresql://username:password@host/database') 6 7# 假设df是包含大量数据的DataFrame 8df.to_sql(name='table_name', con=engine, if_exists='append', index=False, chunksize=1000)
chunksize参数允许你指定每次向数据库提交的数据量，这样可以降低内存压力并加快写入速度。
利用数据库API的批量插入功能
对于大规模数据，直接使用数据库API的批量插入接口可以进一步优化性能：
Python 1import psycopg2 2 3# 连接到PostgreSQL数据库 4conn = psycopg2.connect(database="mydatabase", user="user", password="secret", host="localhost", port="5432") 5cursor = conn.cursor() 6 7# 假设data是一个包含多条记录的二维列表 8for batch in chunks(data, batch_size=1000): # 自定义chunks函数按批分割数据 9 placeholders = ','.join(['%s'] * len(batch[0])) # 创建占位符字符串 10 query = f"INSERT INTO table_name VALUES {placeholders}" 11 cursor.executemany(query, batch) 12 13conn.commit() # 提交事务 14cursor.close() 15conn.close() # 关闭连接
多线程或多进程
对于非常大的数据集，可以利用多线程或多进程并发地批量写入数据：
Python 1import concurrent.futures 2 3def insert_data(batch): 4 # 插入数据的函数，根据实际数据库API调整 5 pass 6 7with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: 8 for data_chunk in divide_chunks(data, 1000): # 将数据分成多个子块 9 executor.submit(insert_data, data_chunk) 10 11# 注意：多线程并不总是加速数据库写入，因为它受到数据库本身的并发处理能力限制 12# 对于支持多连接并发写的数据库（如PostgreSQL），可以适当增加并发数 13# 对于MySQL等数据库，可能需要通过其他方式（如批量大小、串行写入等方式）优化
优化策略
开启事物：在一个大批次的数据写入前开启一个事务，完成后再提交，可以减少数据库的事务开销。
禁用索引：在导入大量数据期间，临时禁用非唯一索引，写入完成后重建索引，可以显著提高写入速度。
配置数据库参数：根据数据库类型调整其写入模式、缓存大小等相关参数，以便更好地应对大量数据写入。
