Python多处理和梳理DFS

Question

我正在将大型数据源读为熊猫，然后将其分解为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还返回正确的数据，但是我不知道如何组合它们，因为多处理正在陷入困境。

Answer 1

上面看起来有些奇怪，也许类似：

from multiprocessing import Pool
SQL = 'SELECT * FROM Clean_SBIR LIMIT %s, %s'
def process_data(offset, limit):
    df = pd.read_sql(SQL, conn, params=(offset, limit))
    return algorithm(df)
with Pool(3) as pool:
   jobs = []
   limit = 1000
   for offset in range(0, 3000, limit):
      jobs.append((offset, limit))
   final_df = pd.concat(pool.starmap(process_data, jobs))

基本上，您正在不必要地复制代码，而不是从块处理算法中返回结果。

也就是说，您可能不想做这样的事情。所有数据都是过程之间的picked，并且是@Serge的一部分。