如何高效用Pandas过滤10GB大CSV并上传Azure Blob？解决耗时与内存问题

核心思路：避免多次扫描源文件 #

原方案耗时70小时的核心问题是多次扫描10GB文件（分块提取邮编后再全局过滤），而map()优化内存爆仓是因为一次性加载了过多数据到内存。最优方向是仅扫描源文件一次，同时完成按邮编的分组与暂存，最后批量上传。

方案1：单循环逐行分组写入临时文件（内存占用极低） #

直接逐行读取CSV，将每行追加到对应邮编的临时文件中，全程仅占用单行数据的内存，完成后统一上传临时文件到Blob。

代码示例 #

import csv
from pathlib import Path
from azure.storage.blob import BlobServiceClient

# 配置Azure Blob连接信息
CONN_STR = "你的Azure存储连接字符串"
CONTAINER_NAME = "目标容器名称"
CSV_PATH = "data/history_{date_to_be_searched}.csv"
POSTAL_CODE_COL = "邮编列的名称"  # 替换为实际列名

# 创建临时目录存放按邮编拆分的文件
temp_dir = Path("./temp_postal_data")
temp_dir.mkdir(exist_ok=True)

# 初始化Blob客户端
blob_service = BlobServiceClient.from_connection_string(CONN_STR)
container = blob_service.get_container_client(CONTAINER_NAME)

# 逐行处理CSV
with open(CSV_PATH, "r", newline="", encoding="utf-8") as csv_in:
    reader = csv.DictReader(csv_in)
    header = reader.fieldnames
    file_cache = {}  # 缓存已打开的文件句柄，避免重复IO

    try:
        for row in reader:
            postal_code = row[POSTAL_CODE_COL].strip()
            if not postal_code:
                continue  # 跳过空邮编行

            # 若该邮编的临时文件未创建，初始化并写入表头
            if postal_code not in file_cache:
                temp_file = temp_dir / f"{postal_code}.csv"
                fh = open(temp_file, "w", newline="", encoding="utf-8")
                writer = csv.DictWriter(fh, fieldnames=header)
                writer.writeheader()
                file_cache[postal_code] = (fh, writer)
            
            # 将当前行写入对应文件
            _, writer = file_cache[postal_code]
            writer.writerow(row)
    finally:
        # 确保所有临时文件被关闭
        for fh, _ in file_cache.values():
            fh.close()

# 批量上传临时文件到Blob
for temp_file in temp_dir.glob("*.csv"):
    postal_code = temp_file.stem
    blob_path = f"filtered/{postal_code}.csv"  # Blob存储路径可自定义
    blob_client = container.get_blob_client(blob_path)
    
    with open(temp_file, "rb") as data:
        blob_client.upload_blob(data, overwrite=True)
    
    # 上传完成后删除本地临时文件
    temp_file.unlink()

优势 #

• 仅扫描源文件1次，耗时直接降低一个数量级（预计几小时内完成）
• 内存占用仅为单行数据大小，完全避免内存溢出
• 逻辑简单，无需复杂的分块或并行处理

方案2：调整分块大小+分块内分组追加（平衡内存与IO） #

如果担心27000+临时文件的IO开销，可以缩小分块尺寸（比如从100万行改为10万行），在每个分块内按邮编分组，再将分组结果追加到对应临时文件（或直接追加到Blob）。

关键优化点 #

• 分块大小调整为10万-50万行（根据内存情况测试，确保单块数据占用内存在100MB以内）
• 每个分块处理时，用字典缓存该块内各邮编的行数据，然后批量追加到对应文件/Blob，而非逐行写入
• 避免全局加载数据，仅保留当前分块的分组结果

额外优化建议 #

1. Azure Blob上传优化：使用upload_blob的chunk_size参数（比如设置为4MB或8MB），提升大文件上传速度；若小文件过多，可考虑将多个邮编文件打包为ZIP后上传（但需符合业务需求）。
2. 异常处理：添加文件写入、Blob上传的异常捕获，避免中途失败前功尽弃。
3. 邮编去重预处理：如果源文件中邮编存在大量重复，可先扫描一次提取所有唯一邮编，提前创建好临时文件的表头，后续逐行写入时无需重复判断表头。

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