parallelize dataframe splitting and processing

Question

Problem statement: How do I parallelize a for loop that splits a pandas dataframe into two parts, applies a function to each part in parallel also, and stores the combined results from the function to a list to use after the loop is over?

For context, I am trying to parallelize my decision tree implementation. Many of the answers I have seen previously related to this question need the result of the function being applied to be a dataframe and the result is just concatenated into a big dataframe. I believe this question is slightly more general.

For example, this is the code I would like to parallelize:

# suppose we have some dataframe given to us
df = pd.DataFrame(....)
computation_results = []
# I would like to parallelize this whole loop and store the results of the
# computations in computation_results. min_rows and total_rows are known
# integers.
for i in range(min_rows, total_rows - min_rows + 1):
    df_left = df.loc[range(0, i), :].copy()
    df_right = df.loc[range(i, total_rows), :].copy()
    # foo is a function that takes in a dataframe and returns some
    # result that has no pointers to the passed dataframe. The following
    # two function calls should also be parallelized.
    left_results = foo(df_left)
    right_results = foo(df_right)
    # combine the results with some function and append that combination
    # to a list. The order of the results in the list does not matter.
    computation_results.append(combine_results(left_results, right_results))
# parallelization is not needed for the following function and the loop is over
use_computation_results(computation_results)

user_2597680 · Answer

Check example in https://docs.python.org/3.3/library/multiprocessing.html#using-a-pool-of-workers.

So in your case:

with Pool(processes=2) as pool:                  # start 2 worker processes
  for i in range(min_rows, total_rows - min_rows + 1):
     df_left = df.loc[range(0, i), :].copy()
     call_left = pool.apply_async(foo, df_left)  # evaluate "foo(df_left)" asynchronously
     df_right = df.loc[range(i, total_rows), :].copy() 
     call_right = pool.apply_async(foo, df_left) # evaluate "foo(df_right)" asynchronously
     left_results = call_left.get(timeout=1)     # wait and get left result
     right_results = call_right.get(timeout=1)   # wait and get right result
     computation_results.append(combine_results(left_results, right_results))

parallelize dataframe splitting and processing

1 Answers