Python Multi-Processing и Combing 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 делал.