Pythonの演算を並列化する

Question

ファイルから条件付き検索を使用して行列を生成しています。ファイルは非常に大きく（25,00,000レコード）、データフレーム（ 'ファイル'）として保存されます。 各行列の行操作（ルックアップ）は、互いに独立しています。とにかくこのプロセスを並列化できますか？

私はpandasとpythonで作業しています。私の現在のアプローチは素朴です。

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

あなたは行列が

で利用可能であるべき

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

を行うことができるはず

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

は私が最速であるかどうかわからないんだけど、あなたの実装よりもはるかに高速である必要があり

Answer 2

[EDIT]ローランドコメントで述べたように、標準のPython実装では、このポストは、CPUのパフォーマンスを改善するために、すべてのソリューションを提供していません。

標準的なPythonの実装では、スレッドはCPUバインドされたタスクで実際にパフォーマンスを改善しません。一度に1つのスレッドだけがPythonバイトコードを実行できるように強制する「グローバルインタープリタロック」があります。これは、メモリ管理の複雑さを抑えるために行われました。

異なる機能に異なるスレッドを使用しようとしましたか？

データフレームを列に分割し、複数のスレッドを作成するとします。次に、各スレッドを割り当てて、関数を列に適用します。あなたは十分な処理能力を持っている場合、あなたは多くの時間を得ることができるかもしれない：250万のレコードの場合

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