このDASKコードの場合並列タスク

Question

一連の発射：私は遅延タスクを実行することでarray内のすべての要素にアクセスしたい

def inc(x): 
    return x + 1 

for x in range(5): 
    array[x] = delay(inc)(x) 

を。しかし、arrayは機能ではないので、array.compute()に電話することはできません。私は

for x in range(5): 
    array[x].compute() 

を行う場合は、各タスクを並列に実行されますんやa[1] GETしかa[0]終了後に発生しますか？このコードを書く良い方法はありますか？

Answer 1

あなたが長い時間がかかるように強制すると、状況が並行して実行されているかどうかを簡単に知ることができます。このコードを実行する場合：

from time import sleep, time 
from dask import delayed 

start = time() 

def inc(x): 
    sleep(1) 
    print('[inc(%s): %s]' % (x, time() - start)) 
    return x + 1 

array = [0] * 5 
for x in range(5): 
    array[x] = delayed(inc)(x) 

for x in range(5): 
    array[x].compute() 

呼び出しが順番に発生することは非常に明白になります。ただし、最後のループを次のように置き換えた場合は、

delayed(array).compute() 

が表示されます。私のマシンでは、出力は次のようになります。

[inc(1): 1.00373506546] 
[inc(4): 1.00429320335] 
[inc(2): 1.00471806526] 
[inc(3): 1.00475406647] 
[inc(0): 2.00795912743] 

明らかに、実行された最初の4つのタスクは並列でした。おそらく、デフォルトの並列処理はマシン上のコアの数に設定されています。なぜなら、CPUを大量に使用するタスクでは、それ以上の処理を行うのは一般的に有用ではないからです。

Answer 2

あなたは一度

from dask import delayed, compute 

array = [delayed(inc)(i) for i in range(5)] 
result = compute(*array)