寻求Python中替代R语言BatchJobs的Slurm集群交互工具

Great question! I’ve dealt with exactly this need before—moving from R’s BatchJobs to Python for Slurm cluster workflows can feel tricky, but there are solid options that cover all your required tasks: submitting batches, checking statuses, retrying failures, and collecting results. Let’s break them down:

1. Joblib + joblib-slurm #

This combo is perfect if you want a simple, familiar interface similar to BatchJobs:

• Submit batch tasks: Define your job function, then use Parallel with the SlurmBackend from joblib-slurm to queue tasks as Slurm jobs. Example snippet:

  from joblib import Parallel, delayed
  from joblib_slurm import SlurmBackend
  
  def my_task(param):
      # Your task logic here
      return result
  
  backend = SlurmBackend(max_jobs=10)
  results = Parallel(backend=backend)(delayed(my_task)(p) for p in my_params)
  

• Check statuses: The backend includes helpers to query job statuses, or you can parse output from Slurm commands like squeue or sacct directly.
• Retry failures: Wrap your task function with a retry decorator (e.g., tenacity) to automatically re-run failed jobs.
• Collect results: Joblib handles gathering outputs from all tasks seamlessly, even across cluster nodes.

2. Dask Distributed + dask-jobqueue #

For scalable, large-scale workflows, Dask is a powerhouse:

• Submit batch tasks: Spin up a Slurm cluster with SLURMCluster from dask-jobqueue, then use client.map() to distribute your tasks across nodes.
• Check statuses: Use the built-in Dask dashboard for real-time monitoring, or programmatically check task statuses via the client object:

  from dask_jobqueue import SLURMCluster
  from dask.distributed import Client
  
  cluster = SLURMCluster(cores=8, memory="16GB")
  cluster.scale(jobs=5)
  client = Client(cluster)
  
  futures = client.map(my_task, my_params)
  # Check status of a specific future
  print(futures[0].status)
  

• Retry failures: Dask has built-in retry support—set retries=N when submitting jobs to automatically re-run failed tasks.
• Collect results: Call client.gather(futures) to pull all results back to your local environment.

3. Custom Wrapper with subprocess #

If you want full control over every step, wrap Slurm’s command-line tools directly:

• Submit jobs: Use subprocess.run() to call sbatch, capture the job ID from the output:

  import subprocess
  
  result = subprocess.run(["sbatch", "my_job_script.sh"], capture_output=True, text=True)
  job_id = result.stdout.strip().split()[-1]
  

• Check statuses: Parse output from sacct -j <job_id> --format=State to determine if jobs are COMPLETED, FAILED, or still running.
• Retry failures: Write a loop that checks job statuses periodically, and re-submit failed jobs using sbatch again.
• Collect results: Use subprocess to run scp (for non-shared filesystems) or directly read result files from a shared cluster filesystem.

Note on PySlurm #

PySlurm is indeed in active development, but it currently lacks some of the high-level workflow helpers that make BatchJobs so convenient. It’s better suited for low-level Slurm API interactions rather than end-to-end batch workflows.

Final Recommendation #

• If you want simplicity close to BatchJobs: Go with joblib + joblib-slurm.
• If you need scalability for large workloads: Choose Dask Distributed + dask-jobqueue.
• If you want full control: Build a custom wrapper with subprocess.

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