如何正确结合ProcessPoolExecutor与DataFrame.GroupBy实现多进程分组计算？

你当前实现卡住的核心原因集中在三点：

• 错误传递GroupBy可迭代对象：df_merged.groupby('ticker')返回的对象内部持有对全量原始DataFrame的引用，主进程向子进程传输单个分组数据时，pickle序列化机制会连带重复携带全表的索引、分组元数据，相当于每处理1家公司就重复序列化一次整个超大DataFrame，所有算力和内存都耗在无效的序列化/反序列化环节，根本没有进入实际业务计算
• 无限制拉满系统资源：默认的ProcessPoolExecutor会创建和CPU核数等量的工作进程，没有预留系统资源做数据传输和结果聚合，当内存占用达到物理上限后会触发系统swap换页，进一步拖慢所有任务执行
• 代码逻辑漏洞：你将executor.map返回的结果迭代器直接放入空集合{}中，没有实际消费计算输出，既无法收集最终结果，也无法感知任务执行进度

核心原则是提前剥离冗余数据，减少进程间的序列化开销，控制进程规模，保留进度反馈能力：

1. 预拆分分组分片：在主进程提前将每个ticker对应的DataFrame拆成独立的、无冗余元数据的小分片，去掉全表带的重复索引，把单分片的序列化体积压到最小
2. 控制工作进程数：一般设置为os.cpu_count() - 1，预留1个CPU核和对应的内存带宽给系统做数据传输、结果拼接，不要100%占满系统资源
3. 消费执行结果：逐一遍历executor.map返回的结果迭代器，同步收集每个分片的计算输出，通过打印日志反馈执行状态

import os
import time
import pandas as pd
from concurrent.futures import ProcessPoolExecutor

# 计算函数必须定义在可被pickle序列化的顶层作用域，避免子进程加载失败
def append_new_company_technicals(group: pd.DataFrame) -> pd.DataFrame:
    # 开启行缓冲打印，避免子进程输出被缓存无法看到执行状态
    ticker = group['ticker'].iloc[0]
    print(f"开始计算 {ticker} 的技术指标", flush=True)

    # 不使用inplace修改，避免引用传递导致的序列化异常
    group = group.sort_values(by='date')
    group['halfvol_30_abs'] = group['px'].rolling(30, min_periods=21).apply(func)
    group['halfvol_180_abs'] = group['px1'].rolling(180, min_periods=135).apply(func)
    group['halfvol_30_rel'] = group['px2'].rolling(30, min_periods=21).apply(func)
    group['halfvol_180_rel'] = group['px3'].rolling(180, min_periods=135).apply(func)

    print(f"完成 {ticker} 的技术指标计算", flush=True)
    return group

if __name__ == "__main__":
    start = time.time()
    # 预拆分分组分片，提前剥离冗余索引
    grouped_shards = [
        g.reset_index(drop=True)
        for _, g in df_merged.groupby('ticker', sort=False)
    ]

    # 控制进程池规模，预留系统资源
    max_workers = os.cpu_count() - 1
    calc_results = []
    with ProcessPoolExecutor(max_workers=max_workers) as executor:
        # 逐次消费map返回的结果，同步收集计算输出
        for shard_result in executor.map(append_new_company_technicals, grouped_shards):
            calc_results.append(shard_result)
    
    # 拼接全量分片为最终结果DataFrame
    final_result_df = pd.concat(calc_results, ignore_index=True)
    end = time.time()
    print(f"多进程计算总耗时: {end - start:.2f} 秒")

• 尽量避免在rolling窗口的自定义apply逻辑里使用纯Python实现，这类逻辑即使多进程拆分也会被进程通信开销抵消收益，可以将计算逻辑改写为numpy向量化实现，从根源缩短单任务耗时
• 如果单个ticker的历史数据分片本身体积超过500MB，不要硬分配给单进程，可以进一步按时间维度拆分任务，避免单分片序列化占用过多内存
• 不要在计算函数里使用inplace=True修改DataFrame，这类操作在子进程的修改不会同步回主进程，还会额外增加数据序列化的开销

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