我是python的新手,我具有為數據計算特征然后返回要處理并寫入文件的列表的功能…我正在使用Pool進行計算,然后使用回調函數來寫入文件,但是沒有調用回調函數,我已經在其中添加了一些print語句,但是絕對沒有調用它.

我的代碼如下所示：

def write_arrow_format(results):

print("writer called")

results[1].to_csv("../data/model_data/feature-"+results[2],sep='',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+"

")

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()

最佳答案

從您的帖子中還不清楚,generate_features返回的列表中包含什么.但是,如果結果1,結果2或文件名中的任何一個不可序列化,則由于某種原因,多處理庫將不會調用回調函數,并且將無法靜默地執行此操作.我認為這是因為多處理庫嘗試在子進程和父進程之間來回傳遞對象之前對其進行酸洗.如果您返回的內容不是“可修復的”(即不可序列化),則不會調用該回調.

我自己遇到了這個錯誤,事實證明這是一個給我麻煩的logger對象的實例.這是一些示例代碼來重現我的問題：

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):

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

instance = ('hello', 'world', ii)

return instance

def pool_test(func):

def callback(item):

print('This is the callback')

print('I have been given the following item: ')

print(item)

num_processes = 3

pool = mp.Pool(processes = num_processes)

results = []

for i in range(5):

res = pool.apply_async(func, (i,), callback=callback)

results.append(res)

pool.close()

pool.join()

def main():

print('#'*30)

print('Calling pool test with bad function')

print('#'*30)

pool_test(bad_test_func)

print('#'*30)

print('Calling pool test with good function')

print('#'*30)

pool_test(good_test_func)

if __name__ == '__main__':

main()

希望這會有所幫助,并為您指明正確的方向.

總結

以上是生活随笔為你收集整理的python apply_async函数_Python-未调用apply_async回调函数的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。