Parallelisieren von Operationen in Python

Question

Ich bevölke eine Matrix mit einer bedingten Suche aus einer Datei. Die Datei ist extrem groß (25.000 Datensätze) und wird als Datenrahmen ("Datei") gespeichert. Jede Matrixzeilenoperation (Lookup) ist unabhängig von der anderen. Kann ich diesen Prozess trotzdem parallelisieren?

Ich arbeite in Pandas und Python. Mein derzeitiger Ansatz ist naiv.

for r in row: 
    for c in column: 
     num=file[(file['Unique_Inventor_Number']==r) & file['AppYearStr']==c)]['Citation'].tolist() 
     num = len(list(set(num))) 
     d.set_value(r, c, num) 

Answer 1

Für 2,5 Millionen Datensätze sollten Sie in der Lage sein

res = file.groupby(['Unique_Inventor_Number', 'AppYearStr']).Citation.nunique() 

der Matrix tun sollten in

verfügbar sein

res.unstack(level=1).fillna(0).values 

ich bin mir nicht sicher, ob das die schnellste ist, sollte aber deutlich schneller sein als Ihre Implementierung

Answer 2

[EDIT] Als Roland in Kommentar erwähnt, in einer Standard-Python-Implementierung, dieser Beitrag keine Lösung anbietet CPU Leistungen zu verbessern.

In der Python-Standardimplementierung verbessern Threads die Leistung von CPU-gebundenen Tasks nicht wirklich. Es gibt eine "globale Interpreter-Sperre", die erzwingt, dass jeweils nur ein Thread Python-Bytecode ausführen kann. Dies wurde getan, um die Komplexität der Speicherverwaltung gering zu halten.

Haben Sie versucht, verschiedene Threads für die verschiedenen Funktionen zu verwenden?

Angenommen, Sie trennen Ihren Datenrahmen in Spalten und erstellen mehrere Threads. Dann weisen Sie jeden Thread zu, um eine Funktion auf eine Spalte anzuwenden. Wenn Sie genügend Rechenleistung haben, können Sie in der Lage sein, viel Zeit zu gewinnen:

from threading import Thread 
import pandas as pd 
import numpy as np 
from queue import Queue 
from time import time 

# Those will be used afterwards 
N_THREAD = 8 
q = Queue() 
df2 = pd.DataFrame() # The output of the script 

# You create the job that each thread will do 
def apply(series, func): 
    df2[series.name] = series.map(func) 


# You define the context of the jobs 
def threader(): 
    while True: 
     worker = q.get() 
     apply(*worker) 
     q.task_done() 

def main(): 

    # You import your data to a pandas dataframe 
    df = pd.DataFrame(np.random.randn(100000,4), columns=['A', 'B', 'C', 'D']) 

    # You create the functions you will apply to your columns 
    func1 = lambda x: x<10 
    func2 = lambda x: x==0 
    func3 = lambda x: x>=0 
    func4 = lambda x: x<0 
    func_rep = [func1, func2, func3, func4] 

    for x in range(N_THREAD): # You create your threads  
     t = Thread(target=threader) 
     t.start() 

    # Now is the tricky part: You enclose the arguments that 
    # will be passed to the function into a tuple which you 
    # put into a queue. Then you start the job by "joining" 
    # the queue 
    for i, func in enumerate(func_rep): 
     worker = tuple([df.iloc[:,i], func]) 
     q.put(worker) 

    t0 = time() 
    q.join() 
    print("Entire job took: {:.3} s.".format(time() - t0)) 

if __name__ == '__main__': 
    main()