在Slurm集群中使用future.batchtools实现R嵌套作业的问题求助

一、将future.batchtools作业转为Slurm数组作业的操作步骤 #

1. 重构R计算脚本 #

修改run_futurebatchtools_job.R，让它能根据Slurm数组任务ID拆分计算逻辑，示例代码如下：

# 加载依赖包
library(future)
library(future.batchtools)

# 获取Slurm数组任务ID，默认值用于本地测试
task_id <- as.integer(Sys.getenv("SLURM_ARRAY_TASK_ID", default = 1))

# 定义所有待执行的计算任务，按需扩展
task_list <- list(
  function() {
    # 任务1的计算逻辑
    Sys.sleep(10)
    return(paste("Task", task_id, "completed"))
  },
  function() {
    # 任务2的计算逻辑
    Sys.sleep(10)
    return(paste("Task", task_id, "completed"))
  }
  # 添加更多任务...
)

# 配置future.batchtools使用Slurm模板
plan(batchtools_slurm, template = "slurm_future_config.tmpl")

# 执行当前数组ID对应的任务
result <- task_list[[task_id]]()

# 保存结果到单独文件，避免冲突
saveRDS(result, file = sprintf("result_task_%d.rds", task_id))

2. 调整Slurm配置模板 #

确保slurm_future_config.tmpl中的资源配置符合子作业需求，示例模板：

#!/bin/bash
#SBATCH --job-name=future_subjob_%j
#SBATCH --output=future_subjob_%j.out
#SBATCH --error=future_subjob_%j.err
#SBATCH --nodes=1
#SBATCH --ntasks=1
#SBATCH --cpus-per-task=2
#SBATCH --mem=8G
#SBATCH --time=01:00:00

# 加载集群R环境（根据实际情况调整）
module load R/4.3.1

# 执行batchtools任务集合
Rscript -e 'batchtools::doJobCollection("{{ job.collection }}")'

3. 编写Slurm数组提交脚本 #

修改run_futurebatchtools.sh，添加数组作业参数，示例：

#!/bin/bash
#SBATCH --job-name=future_array_parent
#SBATCH --output=parent_%A_%a.out
#SBATCH --error=parent_%A_%a.err
#SBATCH --array=1-2  # 数组任务范围，对应task_list的长度
#SBATCH --nodes=1
#SBATCH --ntasks=1
#SBATCH --cpus-per-task=1
#SBATCH --mem=2G
#SBATCH --time=00:10:00

# 加载R环境
module load R/4.3.1

# 运行R脚本
Rscript run_futurebatchtools_job.R

提交命令：sbatch run_futurebatchtools.sh

4. 关键注意事项 #

• 父数组作业仅负责调度，资源配置无需过高，核心计算资源在子作业模板中定义。
• 确认集群允许作业内提交子作业（部分集群需管理员开启权限）。
• 子作业命名规则要统一，方便后续管理。

二、父作业退出后子作业的运行状态 #

future.batchtools底层依赖batchtools，因此会继承父作业退出后子作业仍继续运行的特性。因为子作业是直接提交到Slurm调度系统的独立作业，与父作业的R进程生命周期无关，父进程退出不会影响已提交的子作业。

如果需要父作业退出时终止所有子作业，可以在R脚本中添加退出钩子，示例：

# 在脚本开头注册退出回调
on.exit({
  parent_job_id <- Sys.getenv("SLURM_JOB_ID")
  # 根据子作业命名规则匹配并取消
  subjob_ids <- system(
    sprintf("squeue -h -o %%j -n future_subjob_ | grep %s", parent_job_id),
    intern = TRUE
  )
  if (length(subjob_ids) > 0) {
    system(sprintf("scancel %s", paste(subjob_ids, collapse = " ")))
  }
})

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