如何在多处理完成之前存储所有输出？

Posted 2023-02-16

【中文标题】如何在多处理完成之前存储所有输出？

【英文标题】：How to store all the output before multiprocessing finish?

【发布时间】：2022-01-15 22:28:33

【问题描述】：

我想在 python 中运行多进程。 这是一个例子：

def myFunction(name,age):
     output = paste(name,age)
     return output

names = ["A","B","C"]
ages = ["1","2","3"]

with mp.Pool(processes=no_cpus) as pool:
    results = pool.starmap(myFunction,zip(names,ages))

results_table = pd.concat(results)
results_table.to_csv(file,sep="\t",index=False)

myFunction 在实际情况下需要很长时间。有时我必须中断运行并重新开始。然而，results 只会在所有pool.starmap 完成后写入输出文件。如何在完成之前存储中间/缓存结果？ 我不想将 myFunction 从 return 更改为 .to_csv()

谢谢！

【参考方案1】：

不使用map，而是使用方法imap，它返回一个迭代器，当迭代时，每个结果在它们可用时一一给出（即由my_function返回）。但是，结果仍然必须按顺序返回。如果您不关心订单，请使用imap_unordered。

在返回和迭代每个数据帧时，将其转换为 CSV 文件，并根据是否是正在处理的第一个结果，输出带有或不带有标头的文件。

import pandas as pd
import multiprocessing as mp

def paste(name, age):
    return pd.DataFrame([[name, age]], columns=['Name', 'Age'])

def myFunction(t):
    name, age = t # unpack passed tuple
    output = paste(name, age)
    return output

# Required for Windows:
if __name__ == '__main__':
    names = ["A","B","C"]
    ages = ["1","2","3"]

    no_cpus = min(len(names), mp.cpu_count())

    csv_file = 'test.txt'

    with mp.Pool(processes=no_cpus) as pool:
        # Results from imap must be iterated
        for index, result in enumerate(pool.imap(myFunction, zip(names,ages))):
            if index == 0:
                # First return value
                header = True
                open_flags = "w"
            else:
                header = False
                open_flags = "a"
            with open(csv_file, open_flags, newline='') as f:
                result.to_csv(f, sep="\t", index=False, header=header)

test.txt的输出：

Name    Age
A       1
B       2
C       3

【讨论】：

感谢您的回答。但是我得到了missing 2 required positional arguments: 的imap 错误。我应该提到我有几个论点。

“但是我得到了……”什么？

我刚刚看到您更新的评论。如果您不提供更多信息或发布您的实际代码，或者您现在无法通过阅读imap 上的文档来弄清楚您需要做什么，我应该如何回应？如果你的目标工作函数是一个类的实例方法（例如self.some_name），那么你需要self作为第一个参数，当然，第二个参数代表iterable的每个元素> 被传递给imap，在上面的例子中是tuple。

您好，再次感谢您的回复。在阅读了星图和 imap 后，我现在解决了我的问题。我没有放原始代码的原因是它们很长而且非常具体，也会很混乱。

如果我的回答是您问题的解决方案，那么您应该接受这个答案。