Многопроцессорность в python - совместное использование большого объекта (например, pandas dataframe) между несколькими процессами

Question

Я использую многопроцессорность Python, точнее

from multiprocessing import Pool
p = Pool(15)

args = [(df, config1), (df, config2), ...] #list of args - df is the same object in each tuple
res = p.map_async(func, args) #func is some arbitrary function
p.close()
p.join()

Этот подход имеет огромное потребление памяти; израсходовал почти всю мою оперативную память (в этот момент она становится очень медленной, что делает многопроцессорную работу довольно бесполезной). Я полагаю, проблема в том, что df это огромный объект (большой информационный фрейм pandas), и он копируется для каждого процесса. Я пытался использовать multiprocessing.Value поделиться кадром данных без копирования

shared_df = multiprocessing.Value(pandas.DataFrame, df)
args = [(shared_df, config1), (shared_df, config2), ...] 

(как предложено в многопроцессорной разделяемой памяти Python), но это дает мне TypeError: this type has no size (так же, как разделение сложного объекта между процессами Python?, на который я, к сожалению, не понимаю ответа).

Я впервые использую многопроцессорность, и, возможно, мое понимание (пока) недостаточно хорошее. Является multiprocessing.Value на самом деле даже правильная вещь для использования в этом случае? Я видел другие предложения (например, очередь), но сейчас немного запутался. Какие есть варианты для совместного использования памяти, и какой из них будет лучше в этом случае?

Answer 1

Первый аргумент Value является typecode_or_type. Это определяется как:

typecode_or_type определяет тип возвращаемого объекта: это либо тип ctypes, либо односимвольный тип-тип, используемый модулем массива. * args передается конструктору для типа.

Акцент мой. Таким образом, вы просто не можете поместить pandas dataframe в Value, это должен быть тип ctypes.

Вы могли бы вместо этого использовать multiprocessing.Manager обслуживать ваш экземпляр однокадрового кадра данных всем вашим процессам. Есть несколько разных способов оказаться в одном и том же месте - возможно, самый простой - просто вставить свой фрейм данных в менеджер Namespace,

from multiprocessing import Manager

mgr = Manager()
ns = mgr.Namespace()
ns.df = my_dataframe

# now just give your processes access to ns, i.e. most simply
# p = Process(target=worker, args=(ns, work_unit))

Теперь ваш экземпляр dataframe доступен любому процессу, которому передана ссылка на Manager. Или просто передайте ссылку на Namespaceчище.

Одна вещь, которую я не охватил / не расскажу, это события и сигнализация - если вашим процессам нужно ждать, пока другие завершат выполнение, вам нужно добавить это. Вот страница с некоторыми Event примеры, которые также охватывают чуть более подробно, как использовать менеджер Namespace,

(обратите внимание, что ни один из этих адресов не multiprocessing приведет к ощутимым преимуществам в производительности, это просто даст вам инструменты для изучения этого вопроса)

Answer 2

Вы можете использовать Array вместо того Value для хранения вашего фрейма данных.

Приведенное ниже решение преобразует pandas dataframe в объект, который хранит свои данные в общей памяти:

import numpy as np
import pandas as pd
import multiprocessing as mp
import ctypes

# the origingal dataframe is df, store the columns/dtypes pairs
df_dtypes_dict = dict(list(zip(df.columns, df.dtypes)))

# declare a shared Array with data from df
mparr = mp.Array(ctypes.c_double, df.values.reshape(-1))

# create a new df based on the shared array
df_shared = pd.DataFrame(np.frombuffer(mparr.get_obj()).reshape(df.shape),
                         columns=df.columns).astype(df_dtypes_dict)

Если теперь вы поделитесь df_sharedмежду процессами никаких дополнительных копий делать не будет. Для вашего случая:

pool = mp.Pool(15)

def fun(config):
    # df_shared is global to the script
    df_shared.apply(config)  # whatever compute you do with df/config

config_list = [config1, config2]
res = p.map_async(fun, config_list)
p.close()
p.join()

Это также особенно полезно, если вы используете pandarallel, например:

# this will not explode in memory
from pandarallel import pandarallel
pandarallel.initialize()
df_shared.parallel_apply(your_fun, axis=1)

Примечание: с этим решением вы получите два фрейма данных, которые потребляют вдвое больше памяти и долго инициализируются. Возможно, можно будет читать данные непосредственно в разделяемой памяти.

Answer 3

Вы можете совместно использовать данные pandas между процессами без каких-либо дополнительных затрат памяти, создав дочерний процесс data_handler. Этот процесс получает вызовы от других дочерних элементов с конкретными запросами данных (т. Е. Строкой, определенной ячейкой, срезом и т. Д.) От вашего очень большого объекта фрейма данных. Только процесс data_handler сохраняет ваш фрейм данных в памяти, в отличие от диспетчера, подобного пространству имен, в результате которого фрейм данных копируется во все дочерние процессы. Смотрите ниже рабочий пример. Это может быть преобразовано в бассейн.

Вам нужен индикатор прогресса для этого? смотрите мой ответ здесь: /questions/42641782/pokazat-hod-vyizova-kartyi-mnogoprotsessornogo-pula-python/49414590#49414590

import time
import Queue
import numpy as np
import pandas as pd
import multiprocessing
from random import randint

#==========================================================
# DATA HANDLER
#==========================================================

def data_handler( queue_c, queue_r, queue_d, n_processes ):

    # Create a big dataframe
    big_df = pd.DataFrame(np.random.randint(
        0,100,size=(100, 4)), columns=list('ABCD'))

    # Handle data requests
    finished = 0
    while finished < n_processes:

        try:
            # Get the index we sent in
            idx = queue_c.get(False)

        except Queue.Empty:
            continue
        else:
            if idx == 'finished':
                finished += 1
            else:
                try:
                    # Use the big_df here!
                    B_data = big_df.loc[ idx, 'B' ]

                    # Send back some data
                    queue_r.put(B_data)
                except:
                    pass    

# big_df may need to be deleted at the end. 
#import gc; del big_df; gc.collect()

#==========================================================
# PROCESS DATA
#==========================================================

def process_data( queue_c, queue_r, queue_d):

    data = []

    # Save computer memory with a generator
    generator = ( randint(0,x) for x in range(100) )

    for g in generator:

        """
        Lets make a request by sending
        in the index of the data we want. 
        Keep in mind you may receive another 
        child processes return call, which is
        fine if order isnt important.
        """

        #print(g)

        # Send an index value
        queue_c.put(g)

        # Handle the return call
        while True:
            try:
                return_call = queue_r.get(False)
            except Queue.Empty:
                continue
            else:
                data.append(return_call)
                break

    queue_c.put('finished')
    queue_d.put(data)   

#==========================================================
# START MULTIPROCESSING
#==========================================================

def multiprocess( n_processes ):

    combined  = []
    processes = []

    # Create queues
    queue_data = multiprocessing.Queue()
    queue_call = multiprocessing.Queue()
    queue_receive = multiprocessing.Queue()

    for process in range(n_processes): 

        if process == 0:

                # Load your data_handler once here
                p = multiprocessing.Process(target = data_handler,
                args=(queue_call, queue_receive, queue_data, n_processes))
                processes.append(p)
                p.start()

        p = multiprocessing.Process(target = process_data,
        args=(queue_call, queue_receive, queue_data))
        processes.append(p)
        p.start()

    for i in range(n_processes):
        data_list = queue_data.get()    
        combined += data_list

    for p in processes:
        p.join()    

    # Your B values
    print(combined)


if __name__ == "__main__":

    multiprocess( n_processes = 4 )

Answer 4

По крайней мере, Python 3.6 поддерживает сохранение DataFrame pandas как multiprocessing.Value. См. Рабочий пример ниже:

      import ctypes
import pandas as pd
from multiprocessing import Value

df = pd.DataFrame({'a': range(0,9),
                   'b': range(10,19),
                   'c': range(100,109)})

k = Value(ctypes.py_object)
k.value = df

print(k.value)

Answer 5

Я был очень удивлен, что Parallel от joblib(по крайней мере, начиная с 1.0.1) уже поддерживает совместное использование кадров данных pandas с многопроцессорными рабочими из коробки. По крайней мере, с "loky" бэкендом. Одна вещь, которую я выяснил экспериментально: параметры, которые вы передаете функции, не должны содержать больших dict. Если они это сделают, превратите dict в серию или фрейм данных. Некоторая дополнительная память наверняка используется каждым рабочим, но намного меньше размера вашего предположительно «большого» фрейма данных, находящегося в основном процессе. И вычисление начинается сразу во всех воркерах. В противном случае joblib запускает все запрошенные вами воркеры, но они висят без дела, пока объекты копируются в каждый последовательно, что занимает много времени. Я могу предоставить образец кода, если кому-то это нужно. Я тестировал обработку данных только в режиме только для чтения.