使用Python合并子目录中10万+共200GB CSV文件及数据提取方案咨询

一、合并10万+个CSV文件（200GB） #

处理这类超大规模文件，核心原则是避免一次性加载全部数据到内存，同时优化磁盘IO效率，防止流程中断。

实现思路 #

• 用os.walk遍历所有子目录下的CSV文件
• 逐行读取每个CSV，仅保留第一份文件的表头，后续文件跳过表头重复写入
• 用csv模块做逐行读写，最大限度降低内存占用
• 可选优化：若磁盘IO成为瓶颈，可按子目录分批次处理后再合并，或用多进程（需注意写入时的文件锁问题）

代码实现 #

import os
import csv
from pathlib import Path

def merge_large_csvs(root_dir, output_file):
    header_written = False
    
    # 以写入模式打开输出文件，确保编码统一
    with open(output_file, 'w', newline='', encoding='utf-8') as outfile:
        writer = csv.writer(outfile)
        
        for dirpath, _, filenames in os.walk(root_dir):
            for filename in filenames:
                if filename.lower().endswith('.csv'):
                    file_path = Path(dirpath) / filename
                    print(f"Processing {file_path}")
                    
                    try:
                        with open(file_path, 'r', newline='', encoding='utf-8') as infile:
                            reader = csv.reader(infile)
                            header = next(reader)
                            
                            if not header_written:
                                writer.writerow(header)
                                header_written = True
                            
                            # 逐行写入数据，不占用大量内存
                            for row in reader:
                                writer.writerow(row)
                    except Exception as e:
                        print(f"Error processing {file_path}: {str(e)}")
                        continue

# 调用示例
merge_large_csvs("/path/to/your/csv/root", "merged_total.csv")

关键注意事项 #

• 编码兼容：若CSV存在多种编码（如GBK），可先用chardet库检测编码后再打开，避免乱码
• 内存控制：绝对不要用pandas.read_csv一次性读入，即使分块也要严格控制块大小，防止内存溢出
• 容错机制：保留异常捕获，单个文件出错不会导致整个合并流程终止
• 磁盘空间：确保输出目录有足够存储空间（200GB原始数据合并后体积相近）

二、提取不同文件夹/子文件夹的CSV数据 #

针对12个顶层文件夹、每个含9000个CSV的场景，核心是精准遍历目标路径，按需提取数据（如特定列、符合条件的行）。

实现思路 #

• 遍历指定的12个顶层文件夹，递归处理子目录内的CSV
• 可按顶层文件夹分类保存提取结果，方便后续分析
• 同样采用逐行读取模式，避免内存过载

代码实现（以提取特定列为例） #

import os
import csv
from pathlib import Path

def extract_csv_data(target_dirs, columns_to_extract, output_root):
    Path(output_root).mkdir(exist_ok=True)
    
    for top_dir in target_dirs:
        top_dir_name = Path(top_dir).name
        output_file = Path(output_root) / f"{top_dir_name}_extracted.csv"
        header_written = False
        
        with open(output_file, 'w', newline='', encoding='utf-8') as outfile:
            writer = csv.writer(outfile)
            
            for dirpath, _, filenames in os.walk(top_dir):
                for filename in filenames:
                    if filename.lower().endswith('.csv'):
                        file_path = Path(dirpath) / filename
                        print(f"Extracting from {file_path}")
                        
                        try:
                            with open(file_path, 'r', newline='', encoding='utf-8') as infile:
                                reader = csv.DictReader(infile)
                                # 检查所需列是否存在
                                missing_cols = [col for col in columns_to_extract if col not in reader.fieldnames]
                                if missing_cols:
                                    print(f"Missing columns {missing_cols} in {file_path}, skipping")
                                    continue
                                
                                if not header_written:
                                    writer.writerow(columns_to_extract)
                                    header_written = True
                                
                                # 提取指定列数据
                                for row in reader:
                                    extracted_row = [row[col] for col in columns_to_extract]
                                    writer.writerow(extracted_row)
                        except Exception as e:
                            print(f"Error extracting {file_path}: {str(e)}")
                            continue

# 调用示例：替换为你的12个文件夹路径、目标列和输出目录
target_folders = ["/path/to/folder1", "/path/to/folder2", ...]
extract_csv_data(target_folders, ["user_id", "timestamp", "metric_value"], "./extracted_results")

扩展方向 #

• 若需按条件筛选行（如时间范围过滤），可在逐行读取时加入判断逻辑
• 若需每个CSV单独输出提取结果，可在遍历文件时为每个CSV创建对应输出文件
• 超大规模场景下，可给每个顶层文件夹分配一个进程处理，但要注意磁盘IO的并发压力，避免读写冲突

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