我正在将一个大数据源读入 pandas 并将其分成 3 个块。我想使用多重处理,以便我可以同时完成每个块的分析功能。每个函数之后的输出是一个数据帧。然后我需要组合这三个小数据框。
#This part creates an empty dataframe with the correct column names
d = {'ID': [''], 'Title': [''],'Organization': [''], 'PI': [''],'PI_Phone': [''], 'PI_Email': [''],
'Start_Date': [''], 'End_Date': [''],'FY': [''], 'Funding': [''], 'Abstract': [''],
'URL': [''],'Street': [''], 'City': [''],'State': [''], 'Zip': [''],'Country': ['']}
data = pd.DataFrame(data=d)
def algorithm(df):
print('Alg Running')
df['Abstract'] = df['Abstract'].fillna(value='Abstract')
df['Abstract'] = df['Title'] + ' : ' + df['Abstract']
wide_net = df[df['Abstract'].str.lower().str.contains('|'.join(tissue+te_abstract+temp_abstract+tx_abstract+armi_abstract+['cell ','tissue','organ ']),na=False)]
return wide_net
def chunk1():
print('chunk1')
therange = 0
df1 = pd.read_sql(('SELECT * FROM Clean_SBIR LIMIT {},1000;').format(therange), con=conn)
return algorithm(df1)
def chunk2():
print('chunk2')
therange = 1000
df2 = pd.read_sql(('SELECT * FROM Clean_SBIR LIMIT {},1000;').format(therange), con=conn)
algorithm(df2)
def chunk3():
print('chunk3')
therange = 2000
df3 = pd.read_sql(('SELECT * FROM Clean_SBIR LIMIT {},1000;').format(therange), con=conn)
algorithm(df3)
# creating processes
p1 = multiprocessing.Process(target=chunk1())
p2 = multiprocessing.Process(target=chunk2())
p3 = multiprocessing.Process(target=chunk3())
# starting process 1
p1.start()
# starting process 2
p2.start()
# starting process 3
p3.start()
#This is where I am struggling
results = pd.concat([chunk1(),chunk2(),chunk3()])
# wait until process 1 is finished
p1.join()
# wait until process 2 is finished
p2.join()
# wait until process 3 is finished
p3.join()
print('done')
我的算法函数返回正确的数据,然后 chunk1 也返回正确的数据,但我不知道如何组合它们,因为多处理阻碍了。