Dask分布式磁盘工具报错：无法删除指定临时目录文件

问题详情 #

在Slurm集群上使用Dask进行数据并行处理时，Worker尝试删除指定DASK_TEMPORARY_DIRECTORY下的文件失败，报错信息如下：

distributed.diskutils - ERROR - Failed to remove '/path_to_dask_temporary_directory/scheduler-mwho36kr' (failed in ): [Errno 2] No such file or directory: '/path_to_dask_temporary_directory/scheduler-mwho36kr'

运行Python脚本的Slurm Shell脚本如下：

#!/bin/bash
#SBATCH --job-name=job_name
#SBATCH --nodes=3
#SBATCH --ntasks=3
#SBATCH --cpus-per-task=256
#SBATCH --output=/path_to_log_folder/%x.log

export DASK_TEMPORARY_DIRECTORY=/path_to_dask_temporary_directory

WORKERS=4
THREADS=64

SCHEDULER=scheduler.json
LOGS=/path_to_log_folder

SING_RUN="singularity run --nv -B/directory /path_to_sif_image/$2.sif"

srun --nodes=1 --ntasks=1 --output=$LOGS/%x.schd.log \
      $SING_RUN dask scheduler --scheduler-file $SCHEDULER &> $LOGS/log.schd.out &

srun --nodes=2 --ntasks=2 --output=$LOGS/%x.wrk.log \
      $SING_RUN dask worker --interface ib0 --scheduler-file $SCHEDULER --nworkers=$WORKERS --nthreads=$THREADS &> $LOGS/log.wrk.out &

cd /path_to_python_script &

$SING_RUN python -u $1 &> $LOGS/log.job.out

排查与解决思路 #

1. 确认临时目录的共享性与权限 #

• 确保/path_to_dask_temporary_directory是集群所有节点可访问的共享存储，Slurm多节点任务中，Worker分布在不同节点，若临时目录是本地磁盘，跨节点访问会导致文件找不到。
• 在所有节点执行ls -ld /path_to_dask_temporary_directory，检查运行任务的用户/用户组是否拥有该目录的读、写、执行权限。

2. 调整任务时序与清理逻辑 #

• Slurm可能在主Python任务结束后，提前终止Worker/Scheduler进程，导致Worker后续清理时目标文件已被Scheduler删除或进程被强制终止。修改脚本：
  • 去掉cd /path_to_python_script后的&，避免后续命令在错误目录执行；
  • 在Python脚本执行完成后，添加等待与清理步骤：

    $SING_RUN python -u $1 &> $LOGS/log.job.out
    
    # 等待Scheduler和Worker进程正常退出
    wait
    # 手动清理残留的临时文件（可选）
    rm -rf $DASK_TEMPORARY_DIRECTORY/scheduler-*
    

3. 修复Singularity挂载配置 #

• 检查SING_RUN中的挂载参数，确保DASK_TEMPORARY_DIRECTORY被正确挂载到容器内，否则Worker在容器中无法访问该目录。修改挂载命令：

  SING_RUN="singularity run --nv -B/directory -B$DASK_TEMPORARY_DIRECTORY:$DASK_TEMPORARY_DIRECTORY /path_to_sif_image/$2.sif"
  

4. 调整Dask临时文件清理配置 #

• 在Python脚本中增加清理重试机制，降低因时序问题导致的清理失败概率：

  from dask.distributed import Client
  import dask.config
  
  # 设置清理重试次数和延迟
  dask.config.set({
      "distributed.diskutils.clean-retry": 3,
      "distributed.diskutils.clean-delay": 2
  })
  
  client = Client(scheduler_file="scheduler.json")
  

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