Dask使用远程worker时client.map后调用result/gather崩溃求助

1. 优先排查Worker地址上报异常问题 #

这是Windows跨节点部署Dask的最高频故障点：

• Dask Worker默认会将自身的主机名上报给Scheduler，若两个Windows虚拟机未配置跨节点的主机名解析规则，Scheduler返回给客户端的Worker地址为不可解析的主机名，客户端无法连接Worker拉取任务结果，就会出现lost-data报错，同时Worker侧无异常日志、显示任务执行完成
• 验证方式：在客户端执行print(client.scheduler_info()['workers'])，查看返回的Worker地址是否为ip2的可访问IP，若显示为主机名/内网不可路由IP即可确认问题
• 解决方式：启动ip2的Worker时，新增参数指定上报地址：

  dask worker tcp://10.61.68.34:8786 --host 10.61.68.xx # 替换为ip2的实际IP
  

2. 排查动态端口拦截问题 #

你当前仅放开了8786、8787两个端口，但是Dask Worker与客户端之间的结果传输默认使用随机临时端口，大概率被Windows防火墙拦截：

• 验证方式：临时关闭ip1、ip2两台机器的Windows Defender防火墙，重新运行测试代码，若运行正常即可确认问题
• 解决方式：启动Worker时固定通信端口范围，再将对应端口范围加入防火墙入站规则：

  dask worker tcp://10.61.68.34:8786 --host 10.61.68.xx --worker-port 30000:30200
  

3. 排查版本兼容性问题 #

你使用的2021.9.1版本Dask存在Windows跨节点通信的已知bug，会偶发结果传输时的连接中断：

• 验证方式：将Scheduler、所有Worker、客户端的Dask版本统一升级至2023.3.2及以上，重新运行代码测试
• 注意：所有节点的Dask、Python版本必须完全一致，避免序列化不兼容问题

4. 代码优化建议 #

你当前循环调用每个Future的result()方法会反复建立通信连接，放大网络不稳定的影响，更合理的写法是一次性拉取所有结果：

def do(x):
    y = x**2.0 
    return y
client = Client("tcp://10.61.68.34:8786")
x = client.map(do, range(0, 1799))
r = client.gather(x)
print(r)

内容的提问来源于stack exchange，提问作者Morik-san