Как выводить результаты параллельных вычислений на основе python (ipython-parallel или multiprocessing) в рамку данных pandas?

Question

Простой вопрос: все обучающие программы, которые я прочитал, показывают вам, как выводить результат параллельного вычисления в список (или, в лучшем случае, словарь), используя либо ipython.parallel, либо multiprocessing.

Не могли бы вы указать мне на простой пример вывода результата вычисления в общий кадр данных pandas с использованием библиотек?

http://gouthamanbalaraman.com/blog/distributed-processing-pandas.html - в этом учебном пособии показано, как читать входной фреймворк (код ниже), но тогда как я могу вывести результаты четырех параллельных вычислений в ОДИН информационный кадр, пожалуйста?

import pandas as pd 
import multiprocessing as mp 

LARGE_FILE = "D:\\my_large_file.txt" 
CHUNKSIZE = 100000 # processing 100,000 rows at a time 

def process_frame(df): 
     # process data frame 
     return len(df) 

if __name__ == '__main__': 
     reader = pd.read_table(LARGE_FILE, chunksize=CHUNKSIZE) 
     pool = mp.Pool(4) # use 4 processes 

     funclist = [] 
     for df in reader: 
       # process each data frame 
       f = pool.apply_async(process_frame,[df]) 
       funclist.append(f) 

     result = 0 
     for f in funclist: 
       result += f.get(timeout=10) # timeout in 10 seconds 

     print "There are %d rows of data"%(result) 

Answer 1

Вы спрашиваете multiprocessing (или других питон параллельных модулей) для вывода на структуру данных, что они непосредственно не выводятся. Если вы используете Pool, из любого из параллельных пакетов лучше всего вы получите список (используя map) или итератор (используя imap). Если вы используете общую память от multiprocessing, вы можете получить результат в блок памяти, к которому можно получить доступ через указатель через ctypes.

Итак, вопрос тогда будет, можете ли вы вывести результаты из итератора или из блока разделяемой памяти в pandas.DataFrame? Я думаю, что да. Да, ты можешь. Тем не менее, я не думаю, что я видел простой пример этого в учебнике ... поскольку это не так просто сделать.

Путь итератора кажется гораздо менее вероятным, так как вам нужно будет получить numpy, чтобы переварить итератор, не возвращая результаты обратно в python сначала как список. Я бы пошел с общим маршрутом памяти. Я думаю, что это должно дать вам вывод в DataFrame, который затем можно использовать в multiprocessing:

from multiprocessing import sharedctypes as sh 
from numpy import ctypeslib as ct   
import pandas as pd 

ra = sh.RawArray('i', 4) 
arr = ct.as_array(ra) 
arr.shape = (2,2) 
x = pd.DataFrame(arr) 

Тогда все, что вам придется сделать, это передать дескриптор массива multiprocessing.Process:

import multiprocessing as mp 
p1 = mp.Process(target=doit, args=(arr[:1, :], 1)) 
p2 = mp.Process(target=doit, args=(arr[1:, :], 2)) 
p1.start() 
p2.start() 
p1.join() 
p2.join() 

Затем, с некоторой магией указателя, результат должен быть заполнен вашим DataFrame .

Я оставлю вас, чтобы написать функцию doit, чтобы управлять массивом по своему усмотрению.

EDIT: это похоже на хороший ответ с использованием аналогичного подхода ... https://stackoverflow.com/a/22487898/2379433. Также это также работает: https://stackoverflow.com/a/27027632/2379433.