Pandas分块读取竖线分隔CSV多进程效率低，求最优处理方案

问题核心分析 #

你当前双进程方案效率不升反降的原因很明确：两个进程都在重复读取完整的600MB文件，加上pandas的封装开销、频繁打印的IO消耗，导致资源竞争反而拖慢了速度。要同时优化读取和网络请求效率，需要从「读写分工」「IO复用」两个方向入手。

读取效率优化方案 #

• 替换pandas为标准库csv：pandas的DataFrame带来了大量内存和CPU开销，纯文本CSV处理用csv模块足够，速度能提升30%-50%。
• 合理设置分块大小：将chunksize从500调整为10000-50000（根据内存情况灵活调整），减少文件IO的系统调用次数。
• 进程池分工读取：让每个进程负责读取文件的不同区间，避免重复读文件，最大化磁盘IO利用率。

网络请求效率优化方案 #

• 批量发送请求：将单条数据打包成批量请求（比如每100条发一次），减少HTTP握手和连接建立的开销。
• 异步IO处理请求：用aiohttp实现异步请求，单个进程内可并发处理数十上百个网络请求，比多进程更适合IO密集型场景。
• 读写分离架构：用进程池处理文件读取（CPU/IO混合负载），每个进程内用异步事件循环处理网络请求（纯IO负载），充分利用CPU和网络资源。

综合最优实现代码 #

import time
import csv
from multiprocessing import Pool
import aiohttp
import asyncio

# 配置参数
FILE_PATH = r".\excel.csv"
CHUNK_SIZE = 20000  # 每个进程处理的行数
BATCH_SIZE = 100    # 每批发送的请求数量
ENCODING = "utf-16"
THIRD_PARTY_URL = "https://your-third-party-api.com/endpoint"

def get_file_row_count():
    """获取文件总行数，用于分块"""
    with open(FILE_PATH, "r", encoding=ENCODING) as f:
        return sum(1 for _ in f)

async def send_batch_request(session, batch_data):
    """批量发送请求到第三方接口"""
    # 根据第三方API要求格式化数据，这里只是示例
    payload = {"data": batch_data}
    async with session.post(THIRD_PARTY_URL, json=payload) as resp:
        # 处理响应，比如校验状态码
        if resp.status != 200:
            print(f"Batch request failed with status {resp.status}")

async def process_chunk(chunk_start, chunk_end):
    """处理单个文件块：读取+批量发送请求"""
    batch = []
    async with aiohttp.ClientSession() as session:
        with open(FILE_PATH, "r", encoding=ENCODING) as f:
            reader = csv.reader(f, delimiter="|")
            # 跳过前面的行，定位到当前块的起始位置
            for _ in range(chunk_start):
                next(reader)
            # 读取当前块的内容
            for idx, row in enumerate(reader):
                if idx >= chunk_end - chunk_start:
                    break
                # 将row映射为第三方需要的变量格式，这里示例直接用row
                mapped_data = {"field1": row[0], "field2": row[1]}  # 根据实际字段调整
                batch.append(mapped_data)
                # 达到批量大小则发送请求
                if len(batch) >= BATCH_SIZE:
                    await send_batch_request(session, batch)
                    batch = []
            # 发送剩余的批量数据
            if batch:
                await send_batch_request(session, batch)

def process_chunk_sync(chunk_start, chunk_end):
    """同步包装异步函数，适配进程池"""
    asyncio.run(process_chunk(chunk_start, chunk_end))

if __name__ == "__main__":
    start = time.perf_counter()
    total_rows = get_file_row_count()
    # 拆分文件为多个块
    chunks = [(i, min(i+CHUNK_SIZE, total_rows)) for i in range(0, total_rows, CHUNK_SIZE)]
    
    # 用进程池处理所有块
    with Pool(processes=4) as pool:  # processes设为CPU核心数或略高
        pool.starmap(process_chunk_sync, chunks)
    
    finish = time.perf_counter()
    print(f"Total time: {finish - start:.2f} seconds")

关键优化点说明 #

1. 进程池分块读取：每个进程只读取文件的指定区间，避免重复IO，磁盘利用率最大化。
2. 异步批量请求：单个进程内并发处理多个网络请求，减少等待时间，比多进程更高效处理IO密集任务。
3. 轻量CSV读取：用标准库csv替代pandas，大幅降低内存和CPU开销。
4. 灵活参数配置：可根据服务器性能、第三方API限制调整CHUNK_SIZE和BATCH_SIZE。

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