Dask Distributed：如何删除集群中通过client.upload_file()上传的文件

首先明确结论：dask.distributed 没有原生提供和client.upload_file()功能完全相反的官方接口。client.upload_file()的实现逻辑是将本地文件分发到所有集群Worker的临时工作目录并纳入Python导入路径，Dask本身未内置对应清理逻辑，你可以通过以下替代方案实现需求：

可选实现方法 #

• 方案1：重启集群Worker
  上传的所有文件默认存储在Worker的临时工作目录中，重启Worker节点后临时目录会被系统自动回收清理，之前上传的所有文件都会被直接移除。该方案操作最简单，适合集群运维权限充足、短时间重启不影响业务的场景。

• 方案2：通过client.run()批量执行远程删除
  你可以调用Dask提供的client.run()方法，在所有Worker节点同步执行自定义的文件删除逻辑，示例代码如下：

import os

def delete_remote_file(filename: str):
    file_path = os.path.join(os.getcwd(), filename)
    if os.path.exists(file_path):
        os.remove(file_path)
        return f"Worker删除文件{filename}成功"
    return f"Worker未找到文件{filename}"

# 替换为你实际上传的文件名
exec_result = client.run(delete_remote_file, "your_uploaded_file.py")
# 可打印exec_result查看每个Worker的执行结果

注意：如果上传的是压缩包、目录类文件，需要自行调整删除逻辑适配路径规则

• 方案3：额外清理Python模块缓存
  如果你上传的是可导入的Python模块，删除文件后还可以同步清理Worker上的模块缓存，避免后续导入时命中缓存：

import sys

def unload_remote_module(module_name: str):
    if module_name in sys.modules:
        del sys.modules[module_name]
        return f"Worker卸载模块{module_name}成功"
    return f"Worker未加载模块{module_name}"

# 替换为你实际上传的模块名
client.run(unload_remote_module, "your_module_name")

注意事项 #

• 执行删除操作前请确认文件名准确，避免误删Worker工作目录下的其他任务依赖文件，影响集群正常运行
• 动态伸缩的集群中新拉起的Worker不会包含之前上传的文件，无需额外执行删除操作

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