大数据量获取与存储：规避内存溢出的分批处理方案咨询

阿华AIGC实验室

2026-5-21

方案可行性分析与优化建议

你的思路非常正确——分批加载数据并流式写入CSV是处理大体积数据导出的经典方案，完全能有效控制内存占用，避免溢出问题，可行性很高。下面具体聊聊细节和优化方向：

为什么这个方案可行？

当你每次只拉取1000-2000条记录时，内存中只会保留当前批次的数据对象，处理完成后写入磁盘就可以释放这部分内存，完全能把内存占用控制在1MB左右的目标范围内（只要单条记录的大小不是特别夸张）。这种"分而治之"的思路避开了一次性加载全量数据的内存瓶颈，也是绝大多数大数据导出工具的核心逻辑。

具体优化建议

1. 优化数据库分页查询逻辑

避免使用OFFSET进行分页（比如SELECT * FROM table LIMIT 1000 OFFSET 10000），当偏移量很大时，数据库需要扫描大量无关数据，性能会急剧下降。推荐用基于有序唯一键的游标式分页：

-- 假设id是自增主键，每次查询时传入上一批的最后一个id
SELECT * FROM table WHERE id > last_processed_id ORDER BY id LIMIT 1000;

这种方式数据库可以直接利用索引定位，查询效率稳定，不会随着数据量增大而变慢。

2. 采用流式写入CSV

不要把整批数据先存到内存列表里再一次性写入文件，而是边处理边写。比如在Python中：

import csv
import psycopg2

conn = psycopg2.connect("your_db_config")
cursor = conn.cursor()

last_id = 0
batch_size = 1000

with open('output.csv', 'w', newline='') as csvfile:
    writer = csv.writer(csvfile)
    # 先写入表头
    writer.writerow(['id', 'name', 'email'])
    
    while True:
        cursor.execute("SELECT id, name, email FROM table WHERE id > %s ORDER BY id LIMIT %s", (last_id, batch_size))
        rows = cursor.fetchall()
        if not rows:
            break
        # 写入当前批次数据
        writer.writerows(rows)
        # 更新last_id为当前批次最后一条的id
        last_id = rows[-1][0]
        # 强制刷新到磁盘，避免内存缓存堆积
        csvfile.flush()

cursor.close()
conn.close()

这样每批数据写完就直接刷到磁盘，内存里不会堆积多余数据。