python中令人尴尬的并行问题

Posted 2023-02-16

【中文标题】python中令人尴尬的并行问题

【英文标题】：embarrassingly parallel problem in python

【发布时间】：2022-01-13 08:28:57

【问题描述】：

我有 634 个 *.npy 文件，每个文件都包含一个 2D numpy 形状数组 (8194, 76)。我想以不同的频率对每一列使用 STL 分解五次。所以我想做的是：

for file in files:
    for column in columns:
        for freq in frequencies:
            res = STL(file[:,column], period = freq)
            decomposed = np.vstack((res.trend, res.seasonal, res.resid)).T
    np.save(decompoesd)

最后分解的形状应该是(8194,1140)。我怎样才能使这个并行化？因为串行实现需要 2 个多月的时间。

【问题讨论】：

有什么意见吗？

很抱歉@ArtiomKozyrev 的最新反馈，它通过并行化文件级别完美地工作，非常感谢您的帮助。我做了simulate_cpu_bound(file) 对其进行文件处理并将其写入磁盘。接下来我将尝试并行化 CPU 繁重的任务本身，任何关于如何将每个处理器的 res 输出分组并使用单个进程将其写入磁盘的任何提示，然后只有在此之后，所有其他处理器才会获取下一个文件并执行相同操作跨度>

【参考方案1】：

你可以这样做：

from concurrent.futures import ProcessPoolExecutor


FILES = ["a", "b", "c", "d", "e", "f", "g", "h"]


def simulate_cpu_bound(file):
    2 ** 100000000  # cpu heavy task
    # or just use time.sleep(n), where n - number of seconds
    return file


if __name__ == '__main__':
    with ProcessPoolExecutor(8) as f:
        res = f.map(simulate_cpu_bound, FILES)

    res = list(res)

    print(res)

【讨论】：

multithreading 绝对不是你想要用于 CPU 繁重任务的，除非该任务被实现，例如，作为一个 CPU 语言实现的函数来释放全局解释器锁 (GIL)。您是否尝试在一个循环中计时调用simulate_cpu_bound 8 次，然后为您的多线程版本计时？你不会看到太大的不同。当然，如果您将函数更改为仅sleep 版本，您会因为sleep 不是 CPU 密集型并且确实发布了 GIL。我预计会减少大约 8 倍的时间。

@Booboo 我的错，我什至没有注意导入ThreadPoolExecutor而不是ProcessPoolExecutor，谢谢。我进行了必要的更改。