Dask通过SSHCluster部署集群时futures任务未跨worker并行问题

• 首要原因是Dask默认开启了纯函数结果缓存：你提交的gethost是无参数函数，Dask会默认判定为纯函数，多次提交相同任务时只会实际执行一次，其余任务直接复用第一次的执行结果，因此所有返回值都是同一个worker的主机名。
• 次要原因是Dask默认的调度策略会优先将任务分发到负载更低的节点，你的任务仅包含2秒休眠无CPU消耗，单个12线程的worker足以在调度器将任务分发到第二个worker前执行完所有任务，也会导致结果仅包含单节点主机名。

方案1：禁用任务缓存后提交 #

调用client.submit时添加pure=False参数，关闭纯函数缓存，强制Dask每次提交都独立执行任务：

future=client.submit(gethost, pure=False)

如果要确保任务均衡分发到两个worker，可以调整提交的任务量，或者在提交时指定workers参数手动分配任务到指定节点。

方案2：直接使用client.run获取所有worker执行结果 #

如果你的目标只是获取所有worker节点的主机名，直接调用Dask内置的client.run()方法即可，该方法会自动在所有已连接的worker上执行一次指定函数，无需手动循环提交任务：

# 直接返回所有worker的执行结果，key为worker地址，value为函数返回值
results = client.run(gethost)
print(results)

#!/usr/bin/python3
from dask.distributed import Client, SSHCluster
from time import sleep
import socket
import time

cluster=SSHCluster(["10.20.3.1","c002-interconnect-1","c003-interconnect-1"],connect_options={"known_hosts":None},worker_options={},scheduler_options={"port":0,"dashboard_address":":8788"})
client=Client(cluster)
client.cluster.scale(2)

def gethost():
    sleep(2)
    return socket.gethostname()

start=time.time()
# 直接获取所有worker的主机名
results = client.run(gethost)
end=time.time()

print("workers details:")
for worker_addr, hostname in results.items():
    print(f"worker: {worker_addr}, hostname: {hostname}")
print("time: ")
print(end-start)

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