Dask在分支图中执行重新计算

Question

假设，我创建以下图：

import dask
import time


@dask.delayed
def step_1():
    print("Running Step 1")
    time.sleep(1)
    return True

@dask.delayed
def step_2(prev_step):
    print("Running Step 2")
    time.sleep(1)
    return True

@dask.delayed
def step_3a(prev_step):
    print("Running Step 3a")
    time.sleep(1)
    return True

@dask.delayed
def step_3b(prev_step):
    print("Running Step 3b")
    time.sleep(1)
    return True

stp_1 = step_1()
stp_2 = step_2(stp_1)
stp_3a = step_3a(stp_2)
stp_3b = step_3b(stp_2)

from dask import visualize

visualize([stp_3a, stp_3b])

from dask.distributed import Client, LocalCluster
cluster = LocalCluster(n_workers=1, threads_per_worker=3, dashboard_address="localhost:27998")
client = Client(cluster)
client

​

start = time.perf_counter()

stp_3a_futures = client.compute(stp_3a) # So that the future stays in memory

stp_3a_results = client.gather(stp_3a_futures)

duration = time.perf_counter() - start

print(duration)

[Out]: 3.1600782200694084

这是有道理的。但是现在，当我执行step_3b时，我希望它在一秒钟内完成，因为它已经计算出step_1和step_2。但是，不幸的是，它并不能将这两个步骤保留在内存中，并且step_3b的计算也需要3秒：

start = time.perf_counter()

stp_3b_futures = client.compute(stp_3b) # So that the future stays in memory

stp_3b_results = client.gather(stp_3b_futures)

duration = time.perf_counter() - start

print(duration)

[Out]: 3.0438701044768095

现在，我的问题是：

• 有没有办法保留step_2 and step_1在群集的内存中使用仅 step_3a的延迟对象（即，stp_3a）？

我知道我可以在client.persist（） on stp_2上调用client.persist（），但这不是我想要的答案。在我的用例中，当我将计算step_3a时，我对step_2的延迟对象没有任何引用。

预先感谢您可以回答的人。 :)

Answer 1

Graphain与最近的DASK版本很好地运行：

from time import sleep

from dask import delayed
from dask.config import set as dask_set
from graphchain import optimize


@delayed
def step_1():
    print("Running Step 1")
    sleep(1)
    return True


@delayed
def step_2(prev_step):
    print("Running Step 2")
    sleep(1)
    return True


@delayed
def step_3a(prev_step):
    print("Running Step 3a")
    sleep(1)
    return True


@delayed
def step_3b(prev_step):
    print("Running Step 3b")
    sleep(1)
    return True


stp_1 = step_1()
stp_2 = step_2(stp_1)
stp_3a = step_3a(stp_2)
stp_3b = step_3b(stp_2)

现在，计算：

%time stp_3a.compute()
# Running Step 1
# Running Step 2
# Running Step 3a
# CPU times: user 330 ms, sys: 14.3 ms, total: 344 ms
# Wall time: 3.01 s

%time stp_3b.compute()
# Running Step 1
# Running Step 2
# Running Step 3b
# CPU times: user 6.4 ms, sys: 3.03 ms, total: 9.43 ms
# Wall time: 3.01 s

with dask_set(delayed_optimize=optimize):
    
    %time stp_3a.compute()
    # Running Step 1
    # Running Step 2
    # Running Step 3a
    # CPU times: user 364 ms, sys: 20.8 ms, total: 385 ms
    # Wall time: 3.04 s
    
    %time stp_3b.compute()
    # Running Step 3b
    # CPU times: user 5 ms, sys: 2.97 ms, total: 7.97 ms
    # Wall time: 1.01 s

Answer 2

有（是？）尝试进行一些图形级缓存/回忆，您可能会发现有用：除此之外

​

results = dask.compute([stp_3a, stp_3b])

​将避免重新计算共同的要求。