以下代码可以工作,但由于传递大数据集而非常缓慢。在实际实现中,创建进程和发送数据所需的时间几乎与计算时间相同,因此到创建第二个进程时,第一个进程几乎已完成计算,从而使并行化变得毫无意义。
代码与此问题相同 Multiprocessing has cutoff at 992 integers being joined as result,建议的更改已经实现。然而,我遇到了通常发生的问题,即将大量数据进行pickle处理需要很长时间。
我看到有些答案使用multiprocessing.array来传递共享内存数组。我有一个大约有4000个索引的数组,但每个索引都有一个包含200个键/值对的字典。数据仅由每个进程读取,进行一些计算,然后返回一个矩阵(4000x3)(没有字典)。
像这样的答案Is shared readonly data copied to different processes for Python multiprocessing?使用map。是否可以保持以下系统并实现共享内存?是否有一种有效的方法将数据发送到每个带有字典数组的进程中,例如将字典包装在某个管理器中,然后将其放入multiprocessing.array 中?
代码与此问题相同 Multiprocessing has cutoff at 992 integers being joined as result,建议的更改已经实现。然而,我遇到了通常发生的问题,即将大量数据进行pickle处理需要很长时间。
我看到有些答案使用multiprocessing.array来传递共享内存数组。我有一个大约有4000个索引的数组,但每个索引都有一个包含200个键/值对的字典。数据仅由每个进程读取,进行一些计算,然后返回一个矩阵(4000x3)(没有字典)。
像这样的答案Is shared readonly data copied to different processes for Python multiprocessing?使用map。是否可以保持以下系统并实现共享内存?是否有一种有效的方法将数据发送到每个带有字典数组的进程中,例如将字典包装在某个管理器中,然后将其放入multiprocessing.array 中?
import multiprocessing
def main():
data = {}
total = []
for j in range(0,3000):
total.append(data)
for i in range(0,200):
data[str(i)] = i
CalcManager(total,start=0,end=3000)
def CalcManager(myData,start,end):
print 'in calc manager'
#Multi processing
#Set the number of processes to use.
nprocs = 3
#Initialize the multiprocessing queue so we can get the values returned to us
tasks = multiprocessing.JoinableQueue()
result_q = multiprocessing.Queue()
#Setup an empty array to store our processes
procs = []
#Divide up the data for the set number of processes
interval = (end-start)/nprocs
new_start = start
#Create all the processes while dividing the work appropriately
for i in range(nprocs):
print 'starting processes'
new_end = new_start + interval
#Make sure we dont go past the size of the data
if new_end > end:
new_end = end
#Generate a new process and pass it the arguments
data = myData[new_start:new_end]
#Create the processes and pass the data and the result queue
p = multiprocessing.Process(target=multiProcess,args=(data,new_start,new_end,result_q,i))
procs.append(p)
p.start()
#Increment our next start to the current end
new_start = new_end+1
print 'finished starting'
#Print out the results
for i in range(nprocs):
result = result_q.get()
print result
#Joint the process to wait for all data/process to be finished
for p in procs:
p.join()
#MultiProcess Handling
def multiProcess(data,start,end,result_q,proc_num):
print 'started process'
results = []
temp = []
for i in range(0,22):
results.append(temp)
for j in range(0,3):
temp.append(j)
result_q.put(results)
return
if __name__== '__main__':
main()
解决方法:
只需将字典列表放入管理器中,问题即可得到解决。
manager=Manager()
d=manager.list(myData)
看起来持有列表的经理也管理着该列表包含的字典。启动时间有点慢,因此似乎数据仍在被复制,但这只在一开始完成一次,然后在进程内部对数据进行切片。
import multiprocessing
import multiprocessing.sharedctypes as mt
from multiprocessing import Process, Lock, Manager
from ctypes import Structure, c_double
def main():
data = {}
total = []
for j in range(0,3000):
total.append(data)
for i in range(0,100):
data[str(i)] = i
CalcManager(total,start=0,end=500)
def CalcManager(myData,start,end):
print 'in calc manager'
print type(myData[0])
manager = Manager()
d = manager.list(myData)
#Multi processing
#Set the number of processes to use.
nprocs = 3
#Initialize the multiprocessing queue so we can get the values returned to us
tasks = multiprocessing.JoinableQueue()
result_q = multiprocessing.Queue()
#Setup an empty array to store our processes
procs = []
#Divide up the data for the set number of processes
interval = (end-start)/nprocs
new_start = start
#Create all the processes while dividing the work appropriately
for i in range(nprocs):
new_end = new_start + interval
#Make sure we dont go past the size of the data
if new_end > end:
new_end = end
#Generate a new process and pass it the arguments
data = myData[new_start:new_end]
#Create the processes and pass the data and the result queue
p = multiprocessing.Process(target=multiProcess,args=(d,new_start,new_end,result_q,i))
procs.append(p)
p.start()
#Increment our next start to the current end
new_start = new_end+1
print 'finished starting'
#Print out the results
for i in range(nprocs):
result = result_q.get()
print len(result)
#Joint the process to wait for all data/process to be finished
for p in procs:
p.join()
#MultiProcess Handling
def multiProcess(data,start,end,result_q,proc_num):
#print 'started process'
results = []
temp = []
data = data[start:end]
for i in range(0,22):
results.append(temp)
for j in range(0,3):
temp.append(j)
print len(data)
result_q.put(results)
return
if __name__ == '__main__':
main()
ctypes
对象,然后使用multiprocessing.sharedctypes
。我不确定这是否适用于您的用例。 - danomultiprocessing.Manager.dict
对象放入multiprocessing.sharedctype
对象中,只能放置ctypes
对象。你可以使用Manager
创建一个共享的list
,并在其中放置共享的dict
实例,但我不确定这是否真的会给你带来更好的性能优势。 - dano