python apply_async没有执行函数,未调用apply_async回调函数

我会带着你远行 2022-08-30 06:23 392阅读 0赞

![Image 1][]

I am a newbie to python,i am have function that calculate feature for my data and then return a list that should be processed and written in file.,..i am using Pool to do the calculation and then and use the callback function to write into file,however the callback function is not being call,i ve put some print statement in it but it is definetly not being called.

my code looks like this:

def write\_arrow\_format(results):

print("writer called")

results\[1\].to\_csv("../data/model\_data/feature-"+results\[2\],sep='\\t',encoding='utf-8')

with open('../data/model\_data/arow-'+results\[2\],'w') as f:

for dic in results\[0\]:

feature\_list=\[\]

print(dic)

beginLine=True

for key,value in dic.items():

if(beginLine):

feature\_list.append(str(value))

beginLine=False

else:

feature\_list.append(str(key)+":"+str(value))

feature\_line=" ".join(feature\_list)

f.write(feature\_line+"\\n")

def generate\_features(users,impressions,interactions,items,filename):

\#some processing

return \[result1,result2,filename\]

if \_\_name\_\_=="\_\_main\_\_":

pool=mp.Pool(mp.cpu\_count()-1)

for i in range(interval):

if i==interval:

pool.apply\_async(generate\_features,(users\[begin:\],impressions,interactions,items,str(i)),callback=write\_arrow\_format)

else:

pool.apply\_async(generate\_features,(users\[begin:begin+interval\],impressions,interactions,items,str(i)),callback=write\_arrow\_format)

begin=begin+interval

pool.close()

pool.join()

解决方案

It's not obvious from your post what is contained in the list returned by generate\_features. However, if any of result1, result2, or filename are not serializable, then for some reason the multiprocessing lib will not call the callback function and will fail to do so silently. I think this is because the multiprocessing lib attempts to pickle objects before passing them back and forth between child processes and the parent process. If anything you're returning isn't "pickleable" (i.e not serializable) then the callback doesn't get called.

I've encountered this bug myself, and it turned out to be an instance of a logger object that was giving me troubles. Here is some sample code to reproduce my issue:

import multiprocessing as mp

import logging

def bad\_test\_func(ii):

print('Calling bad function with arg %i'%ii)

name = "file\_%i.log"%ii

logging.basicConfig(filename=name,level=logging.DEBUG)

if ii < 4:

log = logging.getLogger()

else:

log = "Test log %i"%ii

return log

def good\_test\_func(ii):