如何在Dask中以百分比形式获取任务的完成进度？

单个任务场景 #

单个任务只有「未开始、运行中、完成」三种状态，可直接通过状态映射完成百分比：

• 任务状态为finished时，完成百分比为100%
• 任务状态为running时，可自定义为50%（Dask不会提供单任务的细分运行进度，除非在任务内部埋点）
• 任务状态为pending或waiting时，完成百分比为0%

代码示例：

from dask.distributed import Client
import time

def my_task(x):
    time.sleep(5)
    return x ** 2

client = Client()
future = client.submit(my_task, 10, key="my_task_0")

def get_single_task_percent(future):
    status = future.status
    if status == "finished":
        return 100
    elif status == "running":
        return 50
    elif status in ["pending", "waiting"]:
        return 0
    return 0

# 轮询打印进度
while not future.done():
    print(f"当前完成百分比: {get_single_task_percent(future)}%")
    time.sleep(1)
print(f"当前完成百分比: {get_single_task_percent(future)}%")

多任务场景 #

如果是批量任务，可通过「已完成任务数/总任务数」计算精准的完成百分比：

from dask.distributed import Client
import time

def my_task(x):
    time.sleep(5)
    return x ** 2

client = Client()
# 提交10个示例任务
futures = [client.submit(my_task, i, key=f"my_task_{i}") for i in range(10)]

def get_batch_tasks_percent(futures_list):
    total_tasks = len(futures_list)
    if total_tasks == 0:
        return 0
    finished_count = sum(1 for f in futures_list if f.status == "finished")
    return round((finished_count / total_tasks) * 100, 2)

# 轮询打印批量任务进度
while not all(f.done() for f in futures):
    print(f"当前完成百分比: {get_batch_tasks_percent(futures)}%")
    time.sleep(1)
print(f"当前完成百分比: {get_batch_tasks_percent(futures)}%")

注意事项 #

• 单任务场景下，Dask无法提供任务内部的运行进度（比如任务执行到第3秒的百分比），除非你在任务函数中手动埋点上报进度（需要结合Dask的回调机制，复杂度较高）
• 多任务场景的百分比计算是精准的，适合批量任务的进度追踪

内容的提问来源于stack exchange，提问作者dmn