如何在Dask Worker中共享多个文件与文件夹以避免ModuleNotFoundError

你遇到的这个ModuleNotFoundError: No module named 'src.data_processing'是Dask分布式任务里非常常见的问题——因为Dask Workers运行在独立的进程（甚至远程节点）上，它们的Python环境默认不会包含你本地项目的目录结构，所以当任务序列化后传到Worker执行时，找不到对应的模块。下面是几种靠谱的解决办法：

1. 把项目根目录添加到Worker的Python路径（PYTHONPATH） #

这是最直接的方式，确保每个Worker都能找到你的src模块：

• 启动Worker时设置环境变量：
  启动Dask Worker的时候，直接把项目根目录加到PYTHONPATH里：

  PYTHONPATH=/absolute/path/to/your/project/root dask worker tcp://scheduler-address:8786
  

  如果是用dask.distributed.Client在代码里启动Worker，可以通过env参数传递：

  from dask.distributed import Client, Worker
  
  client = Client()
  # 启动Worker时指定PYTHONPATH
  client.start_workers(
      n=2,
      env={"PYTHONPATH": "/absolute/path/to/your/project/root"}
  )
  

• 在代码中动态添加路径（Worker端生效）：
  如果没法在启动时设置，你可以通过client.run让所有Worker执行路径添加操作：

  import sys
  from dask.distributed import Client
  
  client = Client("tcp://scheduler-address:8786")
  
  def add_project_path():
      project_root = "/absolute/path/to/your/project/root"
      if project_root not in sys.path:
          sys.path.append(project_root)
  
  # 让所有Worker执行这个函数
  client.run(add_project_path)
  

2. 直接上传文件/文件夹到所有Worker #

Dask提供了upload_file和upload_dir方法，可以把本地的文件或文件夹直接传到每个Worker的工作目录下，这样Worker就能直接import模块：

from dask.distributed import Client

client = Client("tcp://scheduler-address:8786")
# 上传整个src目录到所有Worker
client.upload_dir("src")

上传完成后，Worker的工作目录里就会有src文件夹，Python解释器能自动识别这个模块。

3. 将项目打包为可安装的Python包 #

如果你的项目是长期维护的，推荐把src目录做成一个标准的Python包（通过setup.py或pyproject.toml配置），然后在所有Worker节点的Python环境中安装这个包：

1. 在项目根目录创建pyproject.toml（示例）：

   [build-system]
   requires = ["setuptools>=61.0"]
   build-backend = "setuptools.build_meta"
   
   [project]
   name = "your-project-name"
   version = "0.1.0"
   packages = ["src"]
   

2. 在每个Worker节点上安装包：

   pip install /path/to/your/project/root
   # 或者如果是Git仓库，用
   pip install git+https://your-repo-url.git
   

这样不管Worker在哪里运行，都能直接import src.data_processing，彻底解决路径问题。

4. 任务序列化时避免直接引用模块（进阶） #

如果你的任务里直接引用了src.data_processing中的函数，pickle序列化时会保存模块路径，Worker找不到就会报错。可以尝试把需要的函数代码内联，或者用cloudpickle替代默认的pickle序列化：

from dask.distributed import Client
import cloudpickle

client = Client("tcp://scheduler-address:8786")
# 设置Dask使用cloudpickle序列化
client.register_serialization("pickle", cloudpickle.dumps, cloudpickle.loads)

不过这种方法只适合简单场景，还是前面几种方法更可靠。

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