97

多くの副問題を含む大きな数値問題を解決しようとしています。Python のマルチプロセッシング モジュール (具体的には Pool.map) を使用して、異なる独立した副問題を異なるコアに分割しています。各サブ問題には、多くのサブサブ問題の計算が含まれます。これらの結果がまだプロセスによって計算されていない場合は、ファイルに保存することでこれらの結果を効果的に記憶しようとしています。それ以外の場合は、計算をスキップしてファイルから結果を読み取るだけです。

ファイルに同時実行性の問題があります。さまざまなプロセスが、サブサブ問題がまだ計算されているかどうかを確認することがあります (結果が保存されるファイルを探すことによって)。計算が実行されていないことを確認します。次に、結果を同時に同じファイルに書き込もうとします。このような書き込み衝突を避けるにはどうすればよいですか?

4

2 に答える 2

158

@ GP89 ​​は良い解決策について言及しました。キューを使用して、ファイルへの唯一の書き込みアクセスを持つ専用プロセスに書き込みタスクを送信します。他のすべてのワーカーには読み取り専用アクセス権があります。これにより、衝突が解消されます。これは apply_async を使用する例ですが、map でも機能します。

import multiprocessing as mp
import time

fn = 'c:/temp/temp.txt'

def worker(arg, q):
    '''stupidly simulates long running process'''
    start = time.clock()
    s = 'this is a test'
    txt = s
    for i in range(200000):
        txt += s 
    done = time.clock() - start
    with open(fn, 'rb') as f:
        size = len(f.read())
    res = 'Process' + str(arg), str(size), done
    q.put(res)
    return res

def listener(q):
    '''listens for messages on the q, writes to file. '''

    with open(fn, 'w') as f:
        while 1:
            m = q.get()
            if m == 'kill':
                f.write('killed')
                break
            f.write(str(m) + '\n')
            f.flush()

def main():
    #must use Manager queue here, or will not work
    manager = mp.Manager()
    q = manager.Queue()    
    pool = mp.Pool(mp.cpu_count() + 2)

    #put listener to work first
    watcher = pool.apply_async(listener, (q,))

    #fire off workers
    jobs = []
    for i in range(80):
        job = pool.apply_async(worker, (i, q))
        jobs.append(job)

    # collect results from the workers through the pool result queue
    for job in jobs: 
        job.get()

    #now we are done, kill the listener
    q.put('kill')
    pool.close()
    pool.join()

if __name__ == "__main__":
   main()
于 2012-11-23T13:38:36.637 に答える
2

Manager結果を一時的にリストに保存してから、リストからファイルに結果を書き込むために使用する必要があるようです。また、starmap処理するオブジェクトと管理リストを渡すために使用します。最初のステップは、 に渡されるパラメータを構築することstarmapです。これには、管理リストが含まれます。

from multiprocessing import Manager
from multiprocessing import Pool  
import pandas as pd

def worker(row, param):
    # do something here and then append it to row
    x = param**2
    row.append(x)

if __name__ == '__main__':
    pool_parameter = [] # list of objects to process
    with Manager() as mgr:
        row = mgr.list([])

        # build list of parameters to send to starmap
        for param in pool_parameter:
            params.append([row,param])

        with Pool() as p:
            p.starmap(worker, params)

この時点から、リストをどのように処理するかを決定する必要があります。大量の RAM と巨大なデータ セットがある場合は、pandas を使用して自由に連結してください。その後、csv または pickle として非常に簡単にファイルを保存できます。

        df = pd.concat(row, ignore_index=True)

        df.to_pickle('data.pickle')
        df.to_csv('data.csv')
于 2016-12-21T17:02:35.040 に答える