Dask-jobqueue

Dask-jobqueue

Dask-jobqueue is a Python library which makes it easy to deploy Dask on common job queuing systems typically found in high performance supercomputers, academic research institutions, and other clusters. Dask is a Python library for parallel computing which scales Python code from multi-core local machines to large distributed clusters in the cloud. Since Dask-jobqueue provides interfaces for OAR and Slurm based clusters, it can be used to facilite the passage between OAR and Slurm based resource managers. Code source, issue and pull requests can be found here.

Installing

pip can be used to install dask-jobqueue and its dependencies:

 pip install dask-jobqueue --upgrade # Install everything from last released version

otherwise, conda can be used to install dask-jobqueue from the conda-forge:

 conda install dask-jobqueue -c conda-forge

Usage

Here is a Python script example which requests for starting a batch script on a well defined resource (2 core, 24GB, at least 1 GPU, specific cluster - chifflet -)

from dask_jobqueue import OARCluster as Cluster
from dask.distributed import Client
import os

cluster = Cluster(
queue='default',
# Should be specified if you belongs to more than one GGA
project='<your grant access group>',
# cores per job, required parameter
cores=2,
# memory per job, required parameter
memory='24GB',
# walltime for each worker job
walltime='1:0:0',

job_extra=[
'-t besteffort',
# reserve node from specific cluster
'-p chifflet',
# reserve node with at least 1 GPU
'-p "gpu_count >= 1"'
],

# another way to reserve node with GPU
resource_spec='gpu=1'
)

client = Client(cluster)

# call your favorite script
client.submit(os.system, "./hello-world.sh").result()

client.close()
cluster.close()

This script can be launched as follow:

flille:

python3 this-script.py

About the resource request, user's configuration can also be specified in ~/.config/dask/jobqueue.yaml file as follow:

jobqueue:
  oar:
    name: dask-worker
    # Dask worker options
    cores: 2
    memory: '24GB'
    processes: 1
    #interface: null # Network interface to use: eth0 or ib0
    death-timeout: 60 # Number of seconds to wait if a worker can not find a scheduler
    #local-directory: null # Location of fast local storage like /scratch or $TMPDIR
    #extra: []
    # OAR resource manager options
    #shebang: "#!/usr/bin/env bash"
    queue: 'default'
    project: null
    walltime: '1:00:00'
    #env-extra: []
    resource-spec: null
    job-extra: []
    log-directory: null
    # Scheduler options
    scheduler-options: {}

The cluster can be then instantiated with one single line as follow:

cluster = OARCluster()

Parameters

dask-jobqueue parameter	OAR command example	Description
queue	#OAR -q	Destination queue for each worker job
project	#OAR --project	Accounting group associated with each worker job
cores	#OAR -l core=2	Total cores per job
memory		Total memory per job
walltime	#OAR -l walltime=hh:mm:ss	Walltime for each worker job
name	#OAR -n	Name of worker, always set to the default value dask-worker
resource_spec	#OAR -l host=1/core=2, gpu=1	Request resources and specify job placement
job_extra	#OAR -O, -E	Log directory
job_extra	#OAR -p parasilo	Property request
job_extra	#OAR -t besteffort	Besteffort job
job_extra	#OAR -r now	Advance reservation
job_extra	#OAR --checkpoint 150	Checkpoint
job_extra	#OAR -a jobid	Jobs dependency

In Dask, a Worker is a Python object and node serving data and performing computations; Jobs are resources submitted to, and managed by, the job queueing system (e.g. OAR, Slurm, etc..).

In Dask-jobqueue, a single Job may include one or more Workers.

The number of Workers can be set by the processes parameter as shown in the example above, if your job can be cut into many processes.

To specify the number of Jobs, you can use the scale command. The number of Jobs can either be specified directly as shown in the example above, or indirectly by the cores or memory request:

# 2 job with 1 worker for each will be launched
cluster.scale(2)
# specify total cores 
cluster.scale(cores=4)
# specify total memory
cluster.scale(memory="48GB")

The cluster generated a traditional job script and submits it number of times as specified to the job queue. You can see the generated job script as follow:

print(cluster.job_script())

The example above will generate the following job script:

#!/usr/bin/env bash

#OAR -n dask-worker
#OAR -q default
#OAR --project mc-staff
#OAR -l gpu=1,walltime=1:0:0
#OAR -O /home/ychi/logs/%jobid%.stdout
#OAR -E /home/ychi/logs/%jobid%.stderr
#OAR -t besteffort
#OAR -p chifflet
#OAR -p "gpu_count >= 1"

/usr/bin/python3 -m distributed.cli.dask_worker tcp://172.16.47.106:39655 --nthreads 1 --nprocs 2 --memory-limit 11.18GiB --name dummy-name --nanny --death-timeout 60 --protocol tcp://

Each of these jobs are sent to the job queue independently. Once the job starts, a dask worker process will start up and connect back to the scheduler running within this process.

All experiment above is tested on Grid5000, OAR based cluster. Plafrim and Cleps are used as Slurm based clusters to run the same experiment, in order to find the common concepts between OAR and Slurm. The table above will be completed with a "Slurm command example" column when Slurm will be fully supported by the Inria's national computing infrastructure.