显示 dask.compute(*something) 调用的进度答案

【问题标题】：Display progress on dask.compute(*something) call显示 dask.compute(*something) 调用的进度
【发布时间】：2021-01-07 20:52:41
【问题描述】：

我使用 Dask 在我的代码中具有以下结构：

@dask.delayed
def calculate(data):
    services = data.service_id
    prices = data.price
    
    return [services, prices]

output = []

for qid in notebook.tqdm(ids):
    r = calculate(parts[parts.quotation_id == qid])
    output.append(r)

事实证明，当我在 output 列表上调用 dask.compute() 方法时，我没有任何进度指示。诊断 UI 不会“捕获”此操作，我什至不确定它是否正常运行（根据我的处理器使用情况判断，我认为不是）。

result = dask.compute(*output)

我正在关注 dask 文档中的“最佳实践”文章：

https://docs.dask.org/en/latest/delayed-best-practices.html

我错过了什么？

编辑：我认为它正在运行，因为我仍然收到内存泄漏/高使用率警告。仍然没有进度指示。

【问题讨论】：

你看到这个帖子了吗：stackoverflow.com/questions/49039750/…
@Val 是的！接受的答案甚至说您可以在仪表板上看到进度（这不会发生）。请注意，仅当我使用我提供的示例中的语法时才会出现此问题（这不是错误的，也不是不好的做法）。谢谢。
所以您在 Jupyter Notebook 中使用“本地”dask（也就是未分发）？编辑：你为什么使用notebook.tqdm？
@Val 是的，我在 jupyter notebook 中本地运行。您是否输入了我在该问题上发布的文档链接？ notebook.tqdm 只是为了获取有关循环的一些信息，它不会改变任何东西。
是的，我很熟悉。我发布了一个为本地和分布式 dask 任务创建进度条的答案。我必须创建一个可重现的示例，所以我希望它与您的实际工作流程相匹配

标签： dask dask-distributed dask-delayed dask-dataframe

【解决方案1】：

正如related post 中所指出的，dask 有两种显示进度的方法：一种用于“正常”dask，另一种用于dask.distributed。

这是一个可重现的例子：

import random 
from time import sleep

import dask
from dask.diagnostics import ProgressBar
from dask.distributed import Client, progress

# simulate work
@dask.delayed
def work(x):

    sleep(x)
    return True
    

# generate tasks

random.seed(42)
tasks = [work(random.randint(1,5)) for x in range(50)]

使用普通的`dask`

ProgressBar().register()
dask.compute(*tasks)

产生：

使用`dask.distributed`

client = Client()
futures = client.compute(tasks)

progress(futures)

产生：

【讨论】：

使用普通的dask

使用dask.distributed

使用普通的`dask`

使用`dask.distributed`