我在使用队列实现两个进程之间的对象(数据帧)交换时遇到了一些问题。
第一个进程从队列中获取数据,第二个进程将数据放入队列中。由于放置数据的进程速度更快,所以获取数据的进程应该通过读取所有对象来清空队列。
我遇到了奇怪的行为,因为我的代码完美地按预期工作,但仅对于100行的数据帧而言,对于1000行的数据帧,获取数据的进程始终仅取得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
任何想法我做错了什么以及如何通过更大的数据框呢?
__version__
: pandas 0.16.2, multiprocessing 0.70a1, python 2.7.10) - chris-scDataFrame
,而是所有超过某个大小阈值的 Python 对象,这将取决于系统。 - chris-sc