Send panda DataFrame's between processes

user_4865723 0
Viewed 94

I do not want to share but just send a DataFrame from one process to another.

The primary DataFrame is cutted into pieces and each piece is processed by a separate process (in the meaning of pythons multiprocessing) on its own CPU core. After the "child" processes are finished they should send back the resulting data frames to concat them together again.

But in my example I reach the END of the script but the DataFrames are never send back via the multiprocessing.Queue object.

#!/usr/bin/env python3
import multiprocessing
import pandas as pd

def worker(df, queue):
    print(multiprocessing.current_process())
    # create new column
    df['X'] = df.b + '-' + df.c
    # modify existing column
    df.d = df.d.apply(lambda x: x.upper())
    # send it back to main process
    queue.put(df)  # tried .copy() also!


if __name__ == '__main__':
    print(pd.__version__)

    # initial data
    df = pd.DataFrame({
        'a': ['A', 'A', 'B', 'B'],
        'b': list('XXXX'),
        'c': list('6218'),
        'd': ['zwei', 'zwei', 'vier', 'neuen']
    })
    # slice the data frame
    df_parts = [
        df.iloc[:2].copy(),
        df.iloc[2:].copy()
    ]

    processes = []

    queue = multiprocessing.Queue()

    for i in range(len(df_parts)):
        p = multiprocessing.Process(target=worker,
                                    args=(df_parts[i], queue) )
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    while queue.full():
        print(queue.get())

    print('END')

I am aware that sending data between multiprocessing.Process instances is pickled in the background. But I do not receive any error about that.

The output:

1.2.5
<Process name='Process-2' parent=14112 started>
<Process name='Process-1' parent=14112 started>
END
>>>
pandas multiprocessing python
1 Answers

You have several problems:

  1. According to the documentation for Queue.full():

Return True if the queue is full, False otherwise. Because of multithreading/multiprocessing semantics, this is not reliable.

So you should not be using this method.

  1. You must never attempt to issue get against a multiprocessig.Queue instance that was written to by a sub-process after you have joined that subprocess. From the documentation:

This means that if you try joining that process you may get a deadlock unless you are sure that all items which have been put on the queue have been consumed. Similarly, if the child process is non-daemonic then the parent process may hang on exit when it tries to join all its non-daemonic children.

  1. You cannot be sure of the order in which the two processes you are creating will be writing their output to the single output queue you have created. If you want to be sure of getting the output in the correct order, create a separate output queue instance for each process. This also simplifies your get processing. If you do want to use one output queue, you know that each process is writing one message and you have N processes, so just issue N get calls and you are done processing the queue.
#!/usr/bin/env python3
import multiprocessing
import pandas as pd

def worker(df, queue):
    print(multiprocessing.current_process())
    # create new column
    df['X'] = df.b + '-' + df.c
    # modify existing column
    df.d = df.d.apply(lambda x: x.upper())
    # send it back to main process
    queue.put(df)  # tried .copy() also!


if __name__ == '__main__':
    print(pd.__version__)

    # initial data
    df = pd.DataFrame({
        'a': ['A', 'A', 'B', 'B'],
        'b': list('XXXX'),
        'c': list('6218'),
        'd': ['zwei', 'zwei', 'vier', 'neuen']
    })
    # slice the data frame
    df_parts = [
        df.iloc[:2].copy(),
        df.iloc[2:].copy()
    ]

    processes = []
    queues = []

    for i in range(len(df_parts)):
        queue = multiprocessing.Queue()
        queues.append(queue)
        p = multiprocessing.Process(target=worker,
                                    args=(df_parts[i], queue) )
        processes.append(p)
        p.start()

    for queue in queues:
        print(queue.get())

    for p in processes:
        p.join()


    print('END')

Prints:

1.3.0
<Process name='Process-1' parent=7748 started>
<Process name='Process-2' parent=7748 started>
   a  b  c     d    X
0  A  X  6  ZWEI  X-6
1  A  X  2  ZWEI  X-2
   a  b  c      d    X
2  B  X  1   VIER  X-1
3  B  X  8  NEUEN  X-8
END

To use one output queue:

    processes = []
    queue = multiprocessing.Queue()

    for i in range(len(df_parts)):
        p = multiprocessing.Process(target=worker,
                                    args=(df_parts[i], queue) )
        processes.append(p)
        p.start()

    for _ in range(len(processes)):
        print(queue.get())

    for p in processes:
        p.join()
user_28237190
Related