pool.map
会在将可迭代对象的部分提交给池中的工作进程之前消耗整个可迭代对象。这就是为什么会出现内存问题。
你应该使用pool.imap
来避免这种情况。关于这一点,请参考此帖子进行详细解释。
话虽如此,我真诚地怀疑多进程处理程序能够像你编写的那样加速程序,因为瓶颈在于磁盘I/O。打开、附加和关闭文件总比一个顺序写操作要慢。并行写入单个文件是不可能的。
假设生成l
需要一些时间,如果你按照以下方式编写程序,则可能会加速:
from contextlib import closing
import multiprocessing
import csv
import pandas as pd
import numpy as np
ref_stops = pd.DataFrame(np.arange(100).reshape((-1, 5)))
nodes = pd.DataFrame(np.arange(400).reshape((-1, 4)))
def haversine(a, b, c, d):
return a*b*c*d
def join_rows(row):
row_list = []
for row2 in ref_stops.itertuples():
row_list.append((row[1], row[2], row[3],
row2[1], row2[2], row2[3], row2[4],
haversine(row[3], row[2], row2[3], row2[2])))
return row_list
def main():
with closing(multiprocessing.Pool()) as pool:
joined_rows = pool.imap_unordered(join_rows, nodes.itertuples(name=None))
with open(r'big_file.csv', 'w') as f:
writer = csv.writer(f)
for row_list in joined_rows:
writer.writerows(row_list)
if __name__ == '__main__':
main()
我猜你不关心顺序,否则你一开始就不会选择多进程了,对吧?这样做的好处是主进程不再生成行列表,而是由工作进程生成。一旦一个工作进程完成了一个列表,它就会将其返回给主进程,然后主进程将其条目附加到文件中。然后工作进程获取新的行并开始构建另一个列表。
在程序中通常使用更多的 pandas 功能可能也会更好(我猜你正在使用 pandas 数据框架,因为有 itertuples
)。例如,你可以创建一个新的数据框架,而不是一个行列表,并使 haversine
与 pandas.Series
对象兼容,这样你就不必在每个条目上调用它。
parama
,nodes
和pool.map(...
返回为空。请添加缺失的代码。 - stovfl