使用 Dataframe 进行多处理和队列

Question

我在通过队列在 2 个进程之间交换对象（数据帧）时遇到了一些麻烦。

第一个进程从队列中获取数据，第二个进程将数据放入队列中。 put-process 速度更快，所以 get-process 应该通过读取所有对象来清除队列。

我的行为很奇怪，因为我的代码运行得很好并且符合预期，但仅适用于数据框中的 100 行，对于 1000 行，get-process 始终只需要 1 个对象。

import multiprocessing, time, sys
import pandas as pd

NR_ROWS = 1000
i = 0
def getDf():
    global i, NR_ROWS
    myheader = ["name", "test2", "test3"]                
    myrow1 =   [ i,  i+400, i+250]
    df = pd.DataFrame([myrow1]*NR_ROWS, columns = myheader)
    i = i+1
    return df 


def f_put(q):
    print "f_put start"        

    while(1): 
        data = getDf()                
        q.put(data)
        print "P:", data["name"].iloc[0]         
        sys.stdout.flush()                    
        time.sleep(1.55)


def f_get(q):
    print "f_get start"    

    while(1):     
        data = pd.DataFrame()

        while not q.empty():
            data = q.get()
            print "get"

        if not data.empty:
            print "G:", data["name"].iloc[0] 
        else:
            print "nothing new"                       
        time.sleep(5.9)


if __name__ == "__main__":

    q = multiprocessing.Queue()

    p = multiprocessing.Process(target=f_put, args=(q,))            
    p.start()
    while(1):
        f_get(q)

    p.join()

100 行数据帧的输出，get-process 获取所有对象

f_get start
nothing new
f_put start
P: 0        # put 1.object into the queue
P: 1        # put 2.object into the queue
P: 2        # put 3.object into the queue
P: 3        # put 4.object into the queue
get         # get-process takes all 4 objects from the queue
get
get
get
G: 3
P: 4
P: 5
P: 6
get
get
get
G: 6
P: 7
P: 8

输出 1000 行数据帧，get-process 只需要一个对象。

f_get start
nothing new
f_put start
P: 0        # put 1.object into the queue
P: 1        # put 2.object into the queue
P: 2        # put 3.object into the queue
P: 3        # put 4.object into the queue
get     <-- #!!! get-process takes ONLY 1 object from the queue!!!
G: 1
P: 4
P: 5
P: 6
get
G: 2
P: 7
P: 8
P: 9
P: 10
get
G: 3
P: 11

知道我做错了什么以及如何传递更大的数据帧吗？

Answer 1

冒着无法完全提供功能齐全的示例的风险，这就是问题所在。

首先是时间问题

我用更大的 DataFrame（10000 甚至 100000）再次尝试了你的代码，我开始看到和你一样的东西。这意味着一旦数组的大小超过某个阈值，您就会看到此行为，该阈值将取决于系统（CPU？）。

我稍微修改了您的代码，以便更容易地看到发生了什么。首先，5 DataFrames 是 put 进入队列，没有任何自定义 time.sleep。在 f_get 函数中，我向循环 (while not q.empty()) 添加了一个计数器（和一个 time.sleep(0)，见下文）。

新代码：

import multiprocessing, time, sys                                                 
import pandas as pd                                                              

NR_ROWS = 10000                                                                  
i = 0                                                                            
def getDf():                                                                     
    global i, NR_ROWS                                                            
    myheader = ["name", "test2", "test3"]                                        
    myrow1 =   [ i,  i+400, i+250]                                               
    df = pd.DataFrame([myrow1]*NR_ROWS, columns = myheader)                      
    i = i+1                                                                      
    return df                                                                    

def f_put(q):                                                                    
    print "f_put start"                                                          
    j = 0                                                                        
    while(j < 5):                                                                
        data = getDf()                                                           
        q.put(data)                                                              
        print "P:", data["name"].iloc[0]                                         
        sys.stdout.flush()                                                       
        j += 1                                                                   

def f_get(q):                                                                    
    print "f_get start"                                                          
    while(1):
        data = pd.DataFrame()                                                    
        loop = 0                                                                 
        while not q.empty():                                                     
            data = q.get()                                                  
            print "get (loop: %s)" %loop
            time.sleep(0)                                         
            loop += 1                                                            
        time.sleep(1.)                                                           

if __name__ == "__main__":                                                       

    q = multiprocessing.Queue()                                                  
    p = multiprocessing.Process(target=f_put, args=(q,))                         
    p.start()                                                                    
    while(1):                                                                    
        f_get(q)                                                                 
    p.join()

现在，如果你运行这对于不同的行数，你会看到这样的东西：

N=100:

f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 2)
get (loop: 3)
get (loop: 4)

N=10000:

f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 0)
get (loop: 0)
get (loop: 0)

这告诉我们什么？ 只要 DataFrame 很小，您关于 put 进程比 get 更快的假设似乎是正确的，我们可以在 while not q.empty() 的一个循环中获取所有 5 个项目。

但是，随着行数的增加，情况发生了变化。 while 条件 q.empty() 的计算结果为 True（队列为空）和外部 while(1) 循环。

这可能意味着 put 现在比 get 慢，我们必须等待。但是，如果我们将整个 f_get 的睡眠时间设置为类似 15 的值，我们仍然会得到相同的行为。

另一方面，如果我们将内部 q.get() 循环中的 time.sleep(0) 更改为 1，

while not q.empty():                                                  
    data = q.get()                                                    
    time.sleep(1)                                                     
    print "get (loop: %s)" %loop                                      
    loop += 1

我们得到这个：

f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 2)
get (loop: 3)
get (loop: 4)

这看起来不错！这意味着实际上 get 做了一些奇怪的事情。看起来当它仍在处理 get 时，队列状态是 empty，并且在 get 完成后下一个项目可用。

我确定这是有原因的，但我对 multiprocessing 不够熟悉，看不出这一点。

根据您的应用程序，您可以将适当的 time.sleep 添加到您的内部循环并查看是否足够。

或者，如果您想解决它（而不是使用变通方法作为 time.sleep 方法），您可以查看 multiprocessing 并查找有关 blocking、非阻塞 或异步通信 - 我认为可以在那里找到解决方案。

使用 Dataframe 进行多处理和队列

Multiprocessing and Queue with Dataframe

python

queue

multiprocessing

pandas

python-multiprocessing