Python多进程场景下如何避免重复读取CSV并高效共享Polars DataFrame？

哈哈，我太懂你这种困惑了——本来想着提前把CSV读好，让多进程共享数据来省时间，结果反而比每个进程各自读还慢，简直反向优化了😂 其实这个问题的核心在于多进程间的数据传递机制，咱们一步步拆解原因，再给你几个可行的解决方案。

为什么直接传DataFrame会变慢？ #

你用的ProcessPoolExecutor在Windows/macOS系统下默认是用spawn模式启动子进程的（Linux默认是fork）。在spawn模式下，子进程会从头开始初始化Python解释器，然后把父进程中需要传递的对象（比如你的DataFrame）通过pickle序列化后传过去。

而Polars的DataFrame底层是基于Arrow格式的大内存块，序列化整个DataFrame的开销特别大——比每个进程单独读一次CSV的开销还要高，这就是为什么你觉得变慢了。

解决方案1：Unix系统下利用写时复制（COW）特性 #

如果你是在Linux或者macOS上跑代码，那可以利用Unix系统的写时复制机制来零开销共享DataFrame。

Unix的fork模式下，子进程会直接共享父进程的内存页，只有当子进程尝试修改内存内容时，才会复制对应的内存页（也就是“写时复制”）。而Polars的DataFrame是不可变对象，你的操作只是过滤生成新的DataFrame，不会修改原DataFrame，所以子进程完全不用复制原数据，直接共享父进程的内存即可。

代码示例（Python 3.10+支持）：

import polars as pl
from concurrent.futures import ProcessPoolExecutor
import multiprocessing

def process_data(df, numbers):
    # 你的处理逻辑，这里只是示例过滤
    return df.filter(pl.col("Points").is_in(numbers)) 

def worker(task_args):
    df, numbers = task_args
    return process_data(df, numbers)

if __name__ == "__main__":
    # 父进程中只读取一次CSV
    df = pl.read_csv("path/to/file.csv")

    # 用fork模式创建进程池，利用写时复制共享DataFrame
    mp_context = multiprocessing.get_context("fork")
    with ProcessPoolExecutor(max_workers=4, mp_context=mp_context) as executor:
        # 把DataFrame和每个任务的参数打包传递
        task_list = [(df, [10, 20]), (df, [30, 40]), (df, [50, 60]), (df, [70, 80])]
        results = list(executor.map(worker, task_list))

这个方案几乎没有额外开销，性能会比每个进程读CSV快很多。

解决方案2：跨平台通用方案——Arrow IPC内存映射 #

如果你的代码需要跨平台（比如要在Windows上跑），那可以用Arrow IPC格式+内存映射的方式来共享数据，这是目前最稳妥的跨平台方案。

Arrow IPC是专门为大数据场景设计的存储格式，支持内存映射读取——简单说就是每个进程不需要把整个数据加载到内存，而是直接“映射”磁盘上的Arrow文件到自己的内存空间，多个进程可以共享这个映射，开销极低。

代码示例：

import polars as pl
from concurrent.futures import ProcessPoolExecutor
import os

def process_data(df, numbers):
    return df.filter(pl.col("Points").is_in(numbers)) 

def worker(numbers):
    # 用内存映射读取Arrow文件，几乎无额外开销
    df = pl.read_ipc("temp_data.arrow", memory_map=True)
    return process_data(df, numbers)

if __name__ == "__main__":
    # 父进程中读取CSV并保存为Arrow IPC临时文件
    df = pl.read_csv("path/to/file.csv")
    df.write_ipc("temp_data.arrow")

    try:
        with ProcessPoolExecutor(max_workers=4) as executor:
            task_params = [[10,20], [30,40], [50,60], [70,80]]
            results = list(executor.map(worker, task_params))
    finally:
        # 任务完成后删除临时文件
        os.remove("temp_data.arrow")

这个方案跨平台通用，性能比重复读CSV好很多，也避免了序列化整个DataFrame的巨大开销。

额外小提醒：别用pickle传Polars DataFrame #

记住，在spawn模式下，绝对不要直接把Polars DataFrame传给子进程——pickle序列化大DataFrame的开销真的会让你怀疑人生。上面的两种方案都是绕开了直接传递DataFrame，要么用内存共享，要么用磁盘映射，这才是高效的正确姿势。

备注：内容来源于stack exchange，提问作者Pedro_Siqueira