Polars如何并行构建计算独立的动态列DataFrame

首先先修正你示例代码的逻辑错误：pl.DataFrame.with_columns 不会原地修改DataFrame，是返回携带新列的新DataFrame对象，你原代码循环里没有接收返回值，执行后根本不会得到新增的指标列。

针对你要并行计算所有独立指标列的需求，不需要自己在Python层实现多线程/多进程并行，Polars本身的Rust计算引擎默认就支持多线程并行调度，只要按正确方式传入表达式，引擎会自动把独立的列计算任务分配到不同CPU核心执行，效率远高于Python层手动实现的并行。

正确实现方式 #

你只需要提前把所有待计算的列表达式收集成列表，一次性传入with_columns即可，Polars会自动完成并行计算：

from typing import List
import polars as pl

# 外部定义的字符串到列生成函数的映射
# 示例mapping结构：
# mapping = {
#     "square": lambda: pl.col("ts_value") ** 2,
#     "square_root": lambda: pl.col("ts_value").sqrt()
# }

class data_frame_constr():
    function_list: List[str]
    time_series: pl.DataFrame

    def compute_indicator_matrix(self) -> pl.DataFrame:
        # 收集所有指标的计算表达式，同时给生成的列设置对应名称
        expr_list = [
            mapping[func_name]().alias(func_name) 
            for func_name in self.function_list
        ]
        # 一次性传入所有表达式，Polars底层自动并行计算独立列
        return self.time_series.with_columns(expr_list)

注意：要触发Polars的原生并行，要求mapping中函数返回的是pl.Expr表达式对象，而不是提前计算好的列值数组。

为什么不推荐Python层手动并行 #

• Polars原生并行没有Python GIL锁限制，也不需要跨进程/线程拷贝原始时间序列数据，没有额外的序列化、通信开销。
• 如果手动用Python的multiprocessing、concurrent.futures做并行，首先需要给每个 worker 拷贝一份完整的原始时间序列数据，数据量稍大时这部分开销就会远超过并行带来的收益，最终速度反而比原生单调用慢。
• 如果你的自定义指标必须用Python逻辑实现（比如用到了map_elements执行Python函数），只需要在写表达式时指定并行策略即可，不需要自己写调度逻辑：

  # 带Python自定义逻辑的指标函数示例
  def complex_mock_indicator():
      return pl.col("ts_value").map_elements(
          lambda x: (x**2 + x**0.5) * 1.2, 
          return_dtype=pl.Float64,
          strategy="threading"  # 开启多线程执行Python函数，绕开GIL限制
      )
  

极端场景备选 #

只有当单个指标的计算逻辑完全无法转化为Polars原生表达式、单指标计算耗时达到秒级以上时，才考虑用多进程做并行计算。但这种场景优先建议把自定义逻辑转化为Polars原生表达式，性能提升通常能达到10~100倍，远高于手动并行的收益。

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