如何在Dask Distributed中访问客户端现有Futures并重新处理结果

问题背景 #

我之前用Dask Distributed提交了1000个模拟任务，代码里的save函数存在bug——本该用open(str(result), "w")保存每个结果到单独文件，结果写成了固定的"result"文件，导致结果被不断覆盖。现在任务已经运行多日，我不想中断进程重新提交，而且当前客户端和调度器的连接还正常，想访问所有已提交的Futures，用正确的逻辑保存现有及后续完成的结果。

解决方案 #

因为客户端连接还正常，我们可以直接从调度器获取所有活跃的Futures，然后用修正后的保存逻辑重新处理已完成和即将完成的任务。具体步骤如下：

1. 连接到现有调度器

不管是在原来的客户端进程里操作，还是新开一个Python脚本，先连接到同一个调度器：

from distributed import Client, as_completed

client = Client("localhost:8786")

2. 获取所有已提交的Futures

通过client.get_futures()方法可以获取调度器上所有的活跃Futures，包括你之前提交的1000个模拟任务：

all_futures = client.get_futures()

如果你原来的客户端进程里还保留着最初的futures变量（就是client.map返回的那个列表），也可以直接用这个变量，效果是一样的。

3. 用修正后的逻辑处理结果

遍历这些Futures，对已完成的任务立即保存结果，对未完成的任务等待其完成后再处理：

def correct_save(result):
    # 修正后的保存函数：每个结果对应单独文件
    with open(str(result), "w") as f:
        print(result, file=f)

# 遍历已完成和待完成的任务
for future in as_completed(all_futures):
    try:
        result = future.result()
        correct_save(result)
        print(f"成功保存结果: {result}")
    except Exception as e:
        # 捕获任务执行失败的情况，避免循环中断
        print(f"任务 {future.key} 执行失败: {str(e)}")

关键注意事项 #

• 如果原来的脚本还在运行那个有bug的save循环，建议先终止它（比如在原来的进程里按下Ctrl+C），避免重复操作或者覆盖文件。
• as_completed会优先处理已经完成的任务，所以你不会错过已经运行完毕的结果，之后会持续等待剩余任务完成并自动保存。
• 加入try-except块是为了防止个别任务失败导致整个处理循环中断，能帮你排查哪些任务出了问题。

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