ProcessPoolExecutor not working for function with multiple arguments - python

user_11154361 0
Viewed 130

I have a dataframe df_full that I am trying to rewrite as a dict() while also doing some stuff over it.

          agent     locations                       modal_choices        
0         agent_1   'loc1', 'loc2', 'loc3', 'loc2'  'mode_1', 'mode_1', 'mode_2', 'mode_3'
1         agent_2   'loc1', 'loc4', 'loc2', 'loc6'  'mode_2', 'mode_3', 'mode_2', 'mode_3'

I am currently facing a problem while trying to multiprocess the following function format_dict() knowing that I only want to iterate over the agent argument, the three others are supposed to be the same for each iterations. So I added the partial() parameter to "freeze" df, dict_ and list_ but the code returns me an empty dict and an empty list by the end and I don't understand why. I suppose I haven't written the executor.map() properly. I tried following the methods shown here but it still doesn't return anything.

What could be wrong with my code?

I also printed the time taken by the following script to run with time.perf_counter() and compared it with what is given with tqdm() but the two values don't match. The iteration part is done in 7 seconds (tqdm) while the print of time.perf_counter() shows up after 2.3 minutes.

What would explain the delay for the ending of the with concurrent.futures.ProcessPoolExecutor() as executor:?

I am, unfortunately, still not an expert in python and this is the first time I'm trying to multiprocess something (as the agent list I am working with is massive and would take days to process...). Any help would be greatly appreciated! And please do tell me if informations are missing or if something is not explained properly, I'll edit the post right away.

def format_dict(agent, df, dict_, list_):
    try:
        dict_[agent] = dict()

        toto_ = df.loc[df.agent_ID == agent]

        toto_mod = toto_['modal_choices'].apply(lambda x: pd.Series(x.split(',')))
        toto_loc = toto_['locations'].apply(lambda x: pd.Series(x.split(',')))

        for i in toto_mod:
            dict_[agent]['step_{}'.format(i)] = dict()
            dict_[agent]['step_{}'.format(i)]['mode']  = toto_mod[i].iloc[0]
            dict_[agent]['step_{}'.format(i)]['start'] = toto_loc[counter + 1].iloc[0]

        dict_[agent]['step_{}'.format(i)]['name']    = dict_agent_edt[agent]['step_0']['name']
        
    except ValueError:
        list_.append(agent)
        
    return dict_, list_


dict_name = dict()
list_name = list()

start = time.perf_counter()

agent = df_full['agent'][:1000]

with concurrent.futures.ProcessPoolExecutor() as executor:
    executor.map(partial(format_dict, df=df_full, dict_=dict_name, list_=list_name),
                 tqdm(agent), chunksize=50)
    
end = time.perf_counter()

print(f'It took {(end-start)/60} minutes.')

Following @Louis Lac's answer, I modified my script to avoid any concurrence but it still returns an empty dict.

def format_dict(agent, df, dict_):
    try:
        dict_[agent] = dict()
        toto_ = df.loc[df.agent_ID == agent]

        (same stuff here)
        
    except ValueError:
        pass
        
    return dict_


start = time.perf_counter()

agents = df_full['agent'][:1000]
dict_name = {}

with concurrent.futures.ProcessPoolExecutor() as executor:
    executor.map(partial(format_dict, df=df_full, dict_=dict_name),
                 tqdm(agents), chunksize=50)

end = time.perf_counter()

print(f'It took {(end-start)/60} minutes.')
pandas dataframe function multiprocessing python
1 Answers

When using concurrency such as multithreading and multiprocessing, functions that are executed concurrently such as format_dict should not mutate shared state to avoid data races or the mutations should be synchronized.

You could for instance compute all your stuff concurrently first, then sequentially reduce the result into outputs (dict_ and list_):

def format_dict(agent, df):
    list_ = None
    try:
        dict_ = dict()

        toto_ = df.loc[df.agent_ID == agent]

        toto_mod = toto_['modal_choices'].apply(lambda x: pd.Series(x.split(',')))
        toto_loc = toto_['locations'].apply(lambda x: pd.Series(x.split(',')))

        for i in toto_mod:
            dict_['step_{}'.format(i)] = dict()
            dict_['step_{}'.format(i)]['mode']  = toto_mod[i].iloc[0]
            dict_['step_{}'.format(i)]['start'] = toto_loc[counter + 1].iloc[0]

        dict_['step_{}'.format(i)]['name'] = dict_agent_edt[agent]['step_0']['name']
        
    except ValueError:
        list_ = agent
        
    return dict_, list_

start = time.perf_counter()

agents = df_full['agent'][:1000]

with concurrent.futures.ProcessPoolExecutor() as executor:
    elements = executor.map(partial(format_dict, df=df_full),
                 tqdm(agents), chunksize=50)
  
dict_ = {}
list_ = []
for agent, (d, l) in zip(agents, elements):
  if l is not None:
    list_.append(l)
  dict_[agent] = d

end = time.perf_counter()

print(f'It took {(end-start)/60} minutes.')
user_63240550
Related