所以我想用下面的代码做的是读取列表的列表并将它们放入名为 checker
的函数中,然后让 log_result
处理结果函数 checker
。我正在尝试使用多线程来执行此操作,因为变量名称 rows_to_parse
实际上有数百万行,因此使用多核应该可以大大加快此过程。
代码目前无法运行并导致 Python 崩溃。
我的顾虑和问题:
- 想要保存在变量
df
中的现有 df 来维护 整个过程中的索引,否则log_result
将得到 对哪一行需要更新感到困惑。 - 我很确定
apply_async
不合适 multiprocessing 函数来执行这个任务,因为我相信 计算机读取和写入 df 的顺序可能会损坏它??? - 我认为可能需要设置一个队列来写入和读取
df
但我不确定我将如何去做。
感谢您的帮助。
import pandas as pd
import multiprocessing
from functools import partial
def checker(a,b,c,d,e):
match = df[(df['a'] == a) & (df['b'] == b) & (df['c'] == c) & (df['d'] == d) & (df['e'] == e)]
index_of_match = match.index.tolist()
if len(index_of_match) == 1: #one match in df
return index_of_match
elif len(index_of_match) > 1: #not likely because duplicates will be removed prior to: if "__name__" == __main__:
return [index_of_match[0]]
else: #no match, returns a result which then gets processed by the else statement in log_result. this means that [a,b,c,d,e] get written to the df
return [a,b,c,d,e]
def log_result(result, dataf):
if len(result) == 1: #
dataf.loc[result[0]]['e'] += 1
else: #append new row to exisiting df
new_row = pd.DataFrame([result],columns=cols)
dataf = dataf.append(new_row,ignore_index=True)
def apply_async_with_callback(parsing_material, dfr):
pool = multiprocessing.Pool()
for var_a, var_b, var_c, var_d, var_e in parsing_material:
pool.apply_async(checker, args = (var_a, var_b, var_c, var_d, var_e), callback = partial(log_result,dataf=dfr))
pool.close()
pool.join()
if __name__ == '__main__':
#setting up main dataframe
cols = ['a','b','c','d','e']
existing_data = [["YES","A","16052011","13031999",3],
["NO","Q","11022003","15081999",3],
["YES","A","22082010","03012001",9]]
#main dataframe
df = pd.DataFrame(existing_data,columns=cols)
#new data
rows_to_parse = [['NO', 'A', '09061997', '06122003', 5],
['YES', 'W', '17061992', '26032012', 6],
['YES', 'G', '01122006', '07082014', 2],
['YES', 'N', '06081992', '21052008', 9],
['YES', 'Y', '18051995', '24011996', 6],
['NO', 'Q', '11022003', '15081999', 3],
['NO', 'O', '20112004', '28062008', 0],
['YES', 'R', '10071994', '03091996', 8],
['NO', 'C', '09091998', '22051992', 1],
['YES', 'Q', '01051995', '02012000', 3],
['YES', 'Q', '26022015', '26092007', 5],
['NO', 'F', '15072002', '17062001', 8],
['YES', 'I', '24092006', '03112003', 2],
['YES', 'A', '22082010', '03012001', 9],
['YES', 'I', '15072016', '30092005', 7],
['YES', 'Y', '08111999', '02022006', 3],
['NO', 'V', '04012016', '10061996', 1],
['NO', 'I', '21012003', '11022001', 6],
['NO', 'P', '06041992', '30111993', 6],
['NO', 'W', '30081992', '02012016', 6]]
apply_async_with_callback(rows_to_parse, df)
最佳答案
在 MultiProcessing 中像这样更新 DataFrame 是行不通的:
dataf = dataf.append(new_row,ignore_index=True)
一方面,这是非常低效的(每次追加的时间复杂度为 O(n),因此总共为 O(n^2)。首选方法是在一次传递中将一些对象连接在一起。
另一方面,更重要的是,dataf 不会为每次更新锁定,因此无法保证两个操作不会发生冲突(我猜这会导致 python 崩溃)。
最后,append 没有执行到位,所以变量 dataf
在回调完成后被丢弃!!并且没有对父 dataf
进行任何更改。
我们可以使用 MultiProcessing list或 dict .如果您不关心顺序,请列出;如果您关心(例如枚举),请列出,因为您必须注意,从异步返回的值不是按照明确定义的顺序返回的。
(或者我们可以自己创建一个实现 Lock 的对象,参见 Eli Bendersky 。)
所以做了如下修改:
df = pd.DataFrame(existing_data,columns=cols)
# becomes
df = pd.DataFrame(existing_data,columns=cols)
d = MultiProcessing.list([df])
dataf = dataf.append(new_row,ignore_index=True)
# becomes
d.append(new_row)
现在,一旦异步完成,您就有了一个 MultiProcessing.list 数据帧。您可以连接这些(和 ignore_index
)以获得所需的结果:
pd.concat(d, ignore_index=True)
应该可以解决问题。
注意:在每个阶段创建新行 DataFrame 的效率也低于让 pandas 一次性将列表列表直接解析为 DataFrame 的效率。希望这是一个玩具示例,真的,您希望您的 block 非常大,以便通过 MultiProcessing 获得胜利(我听说 50kb 作为经验法则......),一次一行永远不会成为一个在这里获胜。
旁白:你应该避免在你的代码中使用全局变量(比如 df),在你的函数中传递它们(在这种情况下,作为检查器的参数)会更干净。
关于python - 多处理写入 Pandas 数据框,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/33550312/