python - 无法将更大的 DataFrames 放入队列中

Question

下面的代码模拟了我面临的多处理问题。

有两个函数 - f1 和 f2 - 将包含 n 行的(pandas)数据帧返回给调用函数 run_fns (n)。这两个函数将并行运行。

对于较小的 n 值(例如 n <= 700)，代码可以正常工作，但对于较大的 n 值(例如 n >= 7000)会卡住。

我尝试使用 Queue([maxsize]) 调用 Queue ，其中包含各种 maxsize 值，包括默认值、0、-1 和许多其他大小不一的数字。改变这种行为。

任何解决方案、变通方法或替代方法都将受到欢迎。我还有一个第二个问题:我真的需要包括

if __name__ == "__main__":

某处？如果有的话在哪里？

代码:f1 返回 n 行和 3 列，f2 返回 n 行和 5 列。数据帧是使用随机生成的整数构建的。

import numpy as np
import pandas as pd
from multiprocessing import Process, Queue


def run_fns(n):
    """Run p1 and p2 in parallel, and get the returned dataframes."""
    q1 = Queue()
    q2 = Queue()
    p1 = Process(target=f1, args=(n, q1))
    p2 = Process(target=f2, args=(n, q2))
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    df1 = q1.get()
    df2 = q2.get()
    return df1, df2


def f1(n, q):
    """Create a dataframe with n rows and 3 columns."""
    df = pd.DataFrame(np.random.randint(n * 3, size=(n, 3)))
    q.put(df)


def f2(n, q):
    """Create a dataframe with n rows and 5 columns."""
    df = pd.DataFrame(np.random.randint(n * 5, size=(n, 5)))
    q.put(df)

Answer 1

您面临一个典型问题，该问题记录在多处理 programming guidelines 中.

Bear in mind that a process that has put items in a queue will wait before terminating until all the buffered items are fed by the “feeder” thread to the underlying pipe. (The child process can call the Queue.cancel_join_thread method of the queue to avoid this behaviour.)

This means that whenever you use a queue you need to make sure that all items which have been put on the queue will eventually be removed before the process is joined. Otherwise you cannot be sure that processes which have put items on the queue will terminate.

您需要确保在加入流程之前获取数据。

# start the processes
p1.start()
p2.start()
# drain the queues
df1 = q1.get()
df2 = q2.get()
# then join the queues
p1.join()
p2.join()

return df1, df2