如何从数据库读取300万条记录并生成拆分CSV文件，支持多线程实现？

安全阿岳

2026-9-28

常见快速实现方案

数据库自带导出工具：比如MySQL的SELECT ... INTO OUTFILE、PostgreSQL的COPY命令，直接按分页条件分批导出，是性能最高的方案之一，全程走数据库原生逻辑，不需要上层应用中转，单库导出的话速度通常能到每秒几万条以上。
批量读取+批量写入的单进程方案：用JDBC/ODBC等驱动的流式读取接口，一次拉取1万~10万条记录到内存，写完一个CSV再拉取下一批，内存占用稳定在百兆以内，实现简单，一般能满足大部分场景的速度要求。
多线程/多进程并行导出：适合数据库性能充足、单线程导出速度遇到IO瓶颈的场景，可以按主键范围、分片字段把数据拆成多段，多个线程同时拉取不同范围的数据分别写CSV，能把导出速度提升数倍。
大数据工具导出：如果是数仓场景，用Spark、Flink等分布式计算工具直接对接数据库分片读取，自动拆分输出多个CSV，适合数据量更大、后续还要做数据处理的场景。

多线程实现步骤

核心逻辑要避免多线程读数据冲突，保证数据不重复不遗漏，优先用范围分片的方式，不要用offset分页（offset过大会导致查询性能极差），具体实现流程如下：

先统计总记录数，计算总分片数：300万条按20万每个文件，总共需要15个CSV，可以直接开15个线程，也可以按CPU核心数开固定线程池，排队处理15个分片任务。
提前给每个分片分配数据范围：比如表主键是自增ID，先查询最小IDmin_id和最大IDmax_id，每个分片的ID范围就是[min_id + i*200000, min_id + (i+1)*200000 -1]，i从0到14。
每个线程独立处理一个分片：线程内部用流式查询拉取对应ID范围的20万条数据，直接写入对应编号的CSV文件，各个线程之间不需要共享数据，也不需要加锁，避免线程安全问题。
所有线程执行完成后，统一校验每个CSV的行数是否符合要求，避免部分分片查询失败导致数据缺失。

以下是Python版本的实现示例：

import pymysql
import csv
from concurrent.futures import ThreadPoolExecutor

# 数据库配置
DB_CONFIG = {
    "host": "你的数据库地址",
    "user": "用户名",
    "password": "密码",
    "database": "库名",
    "charset": "utf8mb4"
}
BATCH_SIZE = 200000
TABLE_NAME = "你的表名"

# 提前获取ID范围
def get_id_range():
    conn = pymysql.connect(**DB_CONFIG)
    cursor = conn.cursor()
    cursor.execute(f"SELECT MIN(id), MAX(id) FROM {TABLE_NAME}")
    min_id, max_id = cursor.fetchone()
    cursor.close()
    conn.close()
    return min_id, max_id

# 单个线程导出逻辑
def export_shard(shard_id, start_id, end_id):
    # 每个线程单独建数据库连接，避免连接共享的线程安全问题
    conn = pymysql.connect(**DB_CONFIG, cursorclass=pymysql.cursors.SSCursor)
    cursor = conn.cursor()
    # 流式查询对应范围的数据
    cursor.execute(f"SELECT * FROM {TABLE_NAME} WHERE id >= %s AND id <= %s", (start_id, end_id))
    # 写CSV
    with open(f"export_{shard_id}.csv", "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        # 写表头
        writer.writerow([i[0] for i in cursor.description])
        # 逐行写入，内存占用极低
        for row in cursor:
            writer.writerow(row)
    cursor.close()
    conn.close()
    print(f"分片{shard_id}导出完成")

if __name__ == "__main__":
    min_id, max_id = get_id_range()
    # 生成所有分片任务
    tasks = []
    current_start = min_id
    shard_id = 1
    while current_start <= max_id:
        current_end = min(current_start + BATCH_SIZE - 1, max_id)
        tasks.append((shard_id, current_start, current_end))
        current_start = current_end + 1
        shard_id += 1
    # 开线程池执行，线程数可根据数据库负载调整
    with ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(lambda x: export_shard(*x), tasks)
    print("所有分片导出完成")

注意：多线程导出的时候不要开太多线程，一般不要超过数据库最大连接数的1/3，否则会把数据库打挂，反而拖慢整体速度。如果你的表没有自增主键，也可以按创建时间、地区等有区分度的字段分片。

内容的提问来源于stack exchange，提问作者user739115