本文介绍在 python 多进程场景下避免重复读取大型 csv 文件的实用方案：通过预切分 + ipc/parquet 序列化替代直接传递 dataframe，显著降低序列化开销并提升整体性能。
本文介绍在 python 多进程场景下避免重复读取大型 csv 文件的实用方案：通过预切分 + ipc/parquet 序列化替代直接传递 dataframe，显著降低序列化开销并提升整体性能。
在使用 concurrent.futures.ProcessPoolExecutor 并行处理大型 CSV 数据时，一个常见误区是将 pl.read_csv() 放入 worker 函数内——这会导致每个子进程重复解析、解码和内存加载，严重浪费 I/O 和 CPU 资源；而另一极端是尝试将完整的 Polars DataFrame 直接传入子进程，看似“只读一次”，实则因 Python 的 pickle 序列化机制（尤其是对底层 Arrow 内存块的深拷贝）引发巨大开销，反而比重复读取更慢。
根本原因在于：Polars DataFrame 在跨进程传递时无法真正共享内存。即使底层基于 Apache Arrow，Python 的 multiprocessing 模块仍需通过 pickle 序列化整个对象（包括其持有的内存缓冲区），导致大量数据被复制、压缩、传输、反序列化，远超一次轻量级二进制文件读取的代价。
✅ 推荐解决方案：预计算 + 高效二进制分片 + 进程本地读取
核心思路是——不在进程间“传递数据”，而是在主进程中预先按任务逻辑切分数据，并以高性能二进制格式（IPC 或 Parquet）持久化为多个小文件；各子进程仅负责读取自身所需的小文件，实现零冗余、低延迟加载。
以下是完整、可运行的优化示例：
使用 qbo-mileage CLI 及用户凭证，从 Airtable、Outlook 或 Google Calendar 记录生成 QuickBooks Online 里程 CSV 文件。
import polars as pl
from concurrent.futures import ProcessPoolExecutor
import os
def process_data(df: pl.DataFrame, numbers: list) -> pl.DataFrame:
"""实际业务逻辑：按 Points 字段过滤并进一步处理"""
return df.filter(pl.col("Points").is_in(numbers)).select(["Name", "Points", "Category"])
def worker(task_param: list) -> pl.DataFrame:
"""子进程入口：从 IPC 文件加载对应分片，执行处理"""
numbers = task_param
filename = f"chunk_{'_'.join(map(str, numbers))}.ipc"
df_chunk = pl.read_ipc(filename)
return process_data(df_chunk, numbers)
# 1. 主进程：一次性读取原始 CSV
print("Step 1: Loading full CSV...")
df_full = pl.read_csv("path/to/file.csv")
# 2. 主进程：按任务参数预切分并写入 IPC 文件（高效、零拷贝序列化）
task_params_list = [[10, 20], [30, 40], [50, 60], [70, 80]]
print("Step 2: Pre-splitting and writing IPC chunks...")
for numbers in task_params_list:
chunk = df_full.filter(pl.col("Points").is_in(numbers))
filename = f"chunk_{'_'.join(map(str, numbers))}.ipc"
chunk.write_ipc(filename)
# 3. 主进程：启动并行处理（各进程独立读取专属 IPC 文件）
print("Step 3: Launching parallel workers...")
with ProcessPoolExecutor(max_workers=4) as executor:
results = list(executor.map(worker, task_params_list))
# 清理临时文件（可选）
for numbers in task_params_list:
filename = f"chunk_{'_'.join(map(str, numbers))}.ipc"
if os.path.exists(filename):
os.remove(filename)
print(f"Done. Got {len(results)} results.")
? 关键优势说明：
- ✅ IPC 格式优先：
write_ipc()/read_ipc()是 Polars 最快的序列化方式，无压缩开销、内存映射友好，特别适合进程间快速加载； - ⚠️ Parquet 备选：若磁盘 I/O 是瓶颈且数据可压缩（如含大量重复字符串），
write_parquet(compression="zstd")可减少磁盘占用，但读取时需解压，需实测权衡； - ? 切分策略灵活：示例按
Points值切分，你可根据业务逻辑改为按行范围（df.slice()）、哈希分桶或时间窗口等，确保各 chunk 独立、负载均衡； - ? 线程安全 & 无共享状态：每个子进程操作完全隔离的文件，规避锁竞争与内存一致性问题。
? 额外优化建议：
- 若
process_data中含 CPU 密集型计算（如复杂表达式、group_by().agg()），可考虑结合pl.Config.set_streaming(True)启用流式处理； - 对超大 CSV（>10GB），建议先用
pl.scan_csv()构建 LazyFrame，再.filter().collect()切分，节省初始内存； - 避免在 worker 中进行
pl.read_csv()—— 即使是小文件，CSV 解析仍比二进制读取慢 3–10 倍。
总结：不要试图“共享” DataFrame，而应“分发”轻量级、预处理好的二进制切片。这一模式兼顾了单次 I/O 成本控制、进程隔离性与 Polars 的极致性能，是生产环境中处理大规模并行分析任务的可靠范式。
