如何用Polars/Pandas高效批量处理CSV数据清洗与转换？

1. 实现并行执行缩短总耗时 #

由于你使用Python 3.11，Ray暂不支持，推荐用标准库的concurrent.futures.ProcessPoolExecutor做进程级并行，适配文件IO+CPU密集型的批量处理场景：

• 核心逻辑：把单文件的读取+清洗封装成独立函数，用进程池分配多进程同时处理不同文件，避开GIL限制
• 代码示例：

import polars as pl
from concurrent.futures import ProcessPoolExecutor

def process_single_file(file_path):
    # 读取CSV并完成清洗
    df = pl.read_csv(file_path)
    # 批量处理所有字符串列（对应第二个问题的优化方法）
    cleaned_df = df.with_columns(
        pl.col(pl.Utf8).str.split(" ").list.eval(
            pl.element().replace("X", None).cast(pl.Float64)
        )
    )
    return cleaned_df

# 替换为你的1000个CSV文件路径列表
file_list = [...]

# 进程数建议设为CPU核心数的1-2倍，避免资源过载
with ProcessPoolExecutor(max_workers=8) as executor:
    results = list(executor.map(process_single_file, file_list))

# 合并所有处理后的DataFrame
final_df = pl.concat(results)

• 内存优化：如果内存压力大，可让每个进程处理完文件后直接写出Parquet中间文件，最后再读取合并，避免同时加载1000个DataFrame到内存。

2. 一次性处理所有列，避免逐列循环 #

Polars方案（推荐，矢量化操作效率更高） #

利用Polars的批量列选择和列表表达式，一次性处理所有字符串列：

cleaned_df = df.with_columns(
    # 选中所有字符串类型的列
    pl.col(pl.Utf8).str.split(" ").list.eval(
        # 对列表内每个元素替换"X"为null，再转float
        pl.element().replace("X", None).cast(pl.Float64)
    )
)

原理：pl.col(pl.Utf8)批量选中所有字符串列，str.split(" ")将字符串分割为列表，list.eval对列表内每个元素执行替换和类型转换，全程是Polars原生矢量化操作，远快于手动逐列循环。

Pandas方案 #

用apply结合批量列选择，虽效率不如Polars，但可避免手动逐列循环：

import pandas as pd

# 选中所有字符串类型的列
str_cols = df.select_dtypes(include="object").columns
df[str_cols] = df[str_cols].apply(
    lambda col: col.str.split(" ").apply(
        lambda lst: [float(x) if x != "X" else pd.NA for x in lst]
    )
)

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