multitraitement en python de partage d'objets volumineux (par exemple, les pandas dataframe) entre plusieurs processus

Je suis à l'aide de Python multitraitement, plus précisément

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()

Cette approche a une énorme consommation de mémoire; manger à peu près tous mes RAM (à quel point il devient extrêmement lent, ce qui rend le multitraitement assez inutile). Je suppose que le problème est que df est un énorme objet (un grand pandas dataframe) et il est copié pour chaque processus. J'ai essayé d'utiliser multiprocessing.Value de partager le dataframe sans copier

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

(comme suggéré dans Python multiprocesseur à mémoire partagée), mais qui me donne TypeError: this type has no size (le même que Le partage d'un objet complexe entre Python processus?, à laquelle malheureusement je ne comprends pas la réponse).

Je suis en utilisant le multitraitement pour la première fois et peut-être que ma compréhension n'est pas (encore) assez bon. Est multiprocessing.Value en fait même la bonne chose à utiliser dans ce cas? J'ai vu d'autres suggestions (par exemple, la file d'attente), mais je suis un peu confus. Quelles options sont là pour partager de la mémoire, et qu'on serait mieux dans ce cas?

  • voir plus récentes question: stackoverflow.com/questions/22468279/....
  • Est-il un récent façon de le faire, ou est l'aide de Namespace la meilleure approche? Comment avez-vous atterri résoudre @Anne
InformationsquelleAutor Anne | 2014-03-18
  1. 28

    Le premier argument de Value est typecode_or_type. Qui est défini comme:

    typecode_or_type détermine le type de l'objet retourné: il est
    un ctypes type ou d'un caractère typecode du type de celui utilisé par
    le tableau module.     *args est transmis au constructeur pour le type.

    C'est moi qui souligne. Donc, vous ne peut tout simplement pas mettre une pandas dataframe dans un Value, il doit être un ctypes type.

    Vous pouvez utiliser un multiprocessing.Manager de servir votre singleton dataframe exemple à l'ensemble de vos processus. Il y a différentes manières de se retrouver dans le même lieu - probablement le plus facile est de simplement plop votre dataframe dans le gestionnaire de 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))

    Maintenant votre dataframe instance est accessible à tout processus qui est passée d'une référence à la Gestionnaire. Ou tout simplement passer une référence à la Namespace, c'est plus propre.

    Une chose que je n'ai pas/ne pas couvrir des événements et de signalisation - si votre processus de besoin d'attendre pour les autres à la fin de l'exécution, vous aurez besoin d'ajouter que, dans. Voici une page avec certains Event exemples qui couvrent un peu plus en détail comment utiliser le manager de Namespace.

    (à noter qu'aucune de ces adresses si multiprocessing est le résultat tangible des avantages de performance, c'est juste de vous donner les outils pour explorer cette question)

    • Merci, cela a fait de la consommation de mémoire beaucoup mieux. Il est encore beaucoup plus élevé que ce que j'aurais pensé si - comment puis-je savoir où la consommation de mémoire est en venir?
    • cette Namespace approche entraîne une grande consommation de mémoire pour moi, trop. J'ai essayé cela avec un DF avec des millions de lignes et 6 colonnes (prise de 2 GO de RAM), et les travailleurs jusqu'à la fin avec ce sujet beaucoup de l'utilisation, trop. Qui plus est, l'accès aux données est rapide (< 1ms) lors de profilés non multitraitement devient vraiment lent pour le travailleur dans le multitraitement contexte. Même après la mem utilisation gonfle au travailleur, une seule ns.df.loc[ix] appel peut prendre plusieurs secondes. @roippi et @Jeff, avez-vous des idées à ce sujet?
    • Tenter cette approche, je peux lire le df à partir de la mémoire partagée, mais je n'étais pas en mesure de changer ses valeurs.
    • J'ai essayé cette approche avec un grand df (chargé à partir d'un ~9go csv), et j'ai fait un formatage de l'erreur qui je suppose est en raison de la taille de la df, "struct.error: 'i' format requires -2147483648 <= number <= 2147483647", des suggestions? Est-il une autre technique?
    • J'ai trouvé que pour les grandes structures de données, de grands dictionnaires ou les pandas dataframes, c'est mieux pour mettre en œuvre cette approche, stackoverflow.com/questions/48464565/...
    InformationsquelleAutor roippi
  2. 0

    Vous pouvez partager une pandas dataframe entre les processus, sans surcharge de la mémoire par la création d'un data_handler processus enfant. Ce processus reçoit des appels de la part d'autres enfants avec les données spécifiques à la demande (c'est à dire une ligne, une cellule spécifique, une tranche etc..) à partir de votre très grande dataframe objet. Seul le data_handler processus maintient votre dataframe dans la mémoire à la différence d'un Gestionnaire comme espace de Noms qui provoque le dataframe à être copié à tous les processus enfants. Voir ci-dessous pour un exemple. Ce peut être converti à la piscine.

    Besoin d'une barre de progression pour cela? voir ma réponse ici: https://stackoverflow.com/a/55305714/11186769

    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 )
    InformationsquelleAutor N.Bell