Dask单Worker进程下任务执行时长异常问题咨询

你的测试结果不符合预期的核心原因是：任务并没有被提交到集群的Worker进程执行，而是在客户端本地的线程池中并行运行了，因此无论多少个任务，总耗时都等于单个任务的4秒。

具体验证与排查步骤 #

1. 检查集群连接状态
   在你的测试代码中添加两行打印，确认客户端是否真的连接到了Worker：

   print("集群可用Worker核心数:", client.ncores())
   print("集群状态:", client)
   

   如果输出的ncores()是空字典{}，说明调度器没有识别到任何Worker，此时Dask会自动切换到本地执行模式，使用客户端机器的CPU核心数创建线程池（比如8核机器会生成8个线程），所以8个任务会并行执行，总耗时4秒。

2. 查看Worker日志
   检查启动Worker的终端日志，确认Worker是否成功连接到调度器。正常连接的日志应该包含：

   distributed.worker - INFO - Connecting to scheduler at tcp://127.0.0.1:8786
   distributed.worker - INFO - Successfully connected to scheduler
   

   如果没有这些日志，说明Worker与调度器的连接失败了。

3. 确认任务执行位置
   在costly_simulation函数中添加进程ID打印，验证任务是在客户端还是Worker进程执行：

   import os
   def costly_simulation(p):
       print(f"任务执行进程ID: {os.getpid()}")
       time.sleep(4)
       return p * 2
   

   如果打印的PID与客户端进程的PID一致，说明任务确实在本地执行。

解决方法 #

1. 确保Worker正确连接到调度器

   • 启动调度器后，复制它输出的完整地址（比如tcp://127.0.0.1:8786），用这个地址启动Worker，避免localhost解析为IPv6地址导致的连接问题：

     dask-worker tcp://127.0.0.1:8786 --nworkers 1 --nthreads 1
     

   • 检查调度器日志，确认是否有Worker连接的记录（正常会显示Register worker相关日志）。

2. 强制任务提交到集群
   提交任务时可以指定workers参数，强制任务发送到集群Worker：

   future = client.submit(costly_simulation, p, workers=['tcp://127.0.0.1:XXXX'])
   

   其中tcp://127.0.0.1:XXXX是Worker的地址，可以从Worker的启动日志中获取。

3. 避免函数序列化问题
   将costly_simulation函数放到单独的Python模块（比如tasks.py）中，然后在客户端和Worker环境中都能导入该模块，这样Dask无需序列化函数，直接在Worker上导入执行，减少潜在的本地执行风险。

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