多处理写入pandas数据帧

Question

所以我想用以下代码做的是读取列表列表并通过调用函数将它们放入checker,然后log_result处理函数的结果checker.我试图使用多线程执行此操作,因为rows_to_parse实际上变量名称有数百万行,因此使用多个核心应该会大幅加快此过程.

目前的代码不起作用并导致Python崩溃.

我有的担忧和问题:

1. 希望变量df中保存的现有df在整个过程中维护索引,否则log_result会对哪行需要更新感到困惑.
2. 我很确定这apply_async不是适当的多处理函数来执行这个任务,因为我相信计算机读取和写入df的顺序可能会破坏它？
3. 我认为可能需要设置一个队列进行写入和读取,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)

Answer 1

在MultiProcessing中更新这样的DataFrames不起作用:

dataf = dataf.append(new_row,ignore_index=True)

一方面,这是非常低效的(每个附加的O(n)所以总共为O(n ^ 2).首选方法是在一次通过中将一些对象连接在一起.

另一方面,更重要的是,dataf没有为每次更新锁定,因此无法保证两个操作不会发生冲突(我猜这是崩溃的python).

最后,append不会就位,所以dataf一旦回调完成就会丢弃变量!! 并且没有对父母进行任何更改dataf.

我们可以使用MultiProcessing list或dict.列表如果您不关心顺序或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)

现在,一旦异步完成,您就拥有了DataFrames的MultiProcessing.list.您可以连接这些(和ignore_index)以获得所需的结果:

pd.concat(d, ignore_index=True)

应该做的伎俩.

注意:在每个阶段创建newrow DataFrame的效率也低于让pandas一次性将列表列表直接解析为DataFrame的效率.希望这是一个玩具示例,你真的希望你的大块能够获得MultiProcessing的胜利(我听说50kb作为一个经验法则......),一次一行永远不会是一个在这里赢.

旁白:你应该避免在代码中使用全局变量(如df),在函数中传递它们会更加清晰(在这种情况下,作为checker的参数).