Dask-Jobqueue unable to run on compute node

[JatinGharat94]

I am trying to run some computation on cray compute nodes using PBSCluster.
I have written the dask-jobqueue script as follows:

import xarray as xr
from dask.distributed import Client
from dask_jobqueue import PBSCluster
from dask import compute
import dask

cluster = PBSCluster(queue='research',
                     project='DaskOnPBS',
                     local_directory='/hpcs_rnd/Python_Data_Analysis/Jatin/dask-worker-space/',
                     cores=1,
                     processes=1,
                     memory='10GB',
                     resource_spec='select=1:ncpus=36:mem=20GB:vntype=cray_compute')

cluster.scale(8)
client = Client(cluster)

But IT administrators told me that my script runs on login node rather than on compute node. They told me to run this script on compute node using aprun command which is similar to mpirun.

Usually job scripts are submitted on our HPC in the following manner:

#!/bin/sh
#PBS -N any_job
#PBS -q research
#PBS -l select=1:ncpus=40:vntype=cray_compute
#PBS -l walltime=500:00:00

cd $WORK_DIRECTORY
aprun -n 40 -N 40 ./your_filename > output.log

I do not know how to use aprun command in PBSCluster to run my script on compute node. Please help!

[guillaumeeb]

Hi @JatinGharat94, did you find the solution to your problem?

It looks like you thought using n_workers kwarg would start several worker jobs, but this kwarg is not supported by dask-jobqueue. Please call scale method on your cluster object if you want to start workers.

[lesteve]

But IT administrators told me that my script runs on login node rather than on compute node.

I don't know anything about PBS and have never heard of aprun before. Having said that I think your question is fairly general.

Say you have a script like this (main-dask-script.py):

# main-dask-script.py
cluster = PBSCluster(...)

cluster.scale(jobs=1)
client = Client(cluster)
fut = client.submit(lambda x: x + 1, 10)
print('Result:', fut.result())

You have mostly three options to run it on a cluster, from the simplest to the most complicated:

Option 1: on the login node

This assumes you can run this on the login node.

python main-dask-script.py

In @JatinGharat94's case IT seems not happy with this option. In practice, as long as you do most of your computation on the Dask workers main-dask-script.py should not consume too many resources but I agree that this is probably not aligned with cluster best practices.

Option 2: Interactive job option

Easier for testing: start an interactive job (check your cluster doc or ask IT). Once in your interactive job, you are on a compute node and you can run your main Python script in your interactive job the same way as Option 1.

Note: depending on your cluster configuration, interactive queues may have short walltime so this is not feasible when main-dask-script.py takes a long time. This is one of the reason why you may want to go to Option 3.

Option 3 Submitting the main Dask script as a job

Better for "production" i.e. when your main script is fully debugged.

# you probably need some options here for using the right queue
bsub main-dask-script.py

Warning: if your job where the main-dask-script.py runs hits its walltime, you will lose all your work (unless you have some mechanism to save part of your computation): the Dask scheduler lives in main-dask-script.py and the Dask workers will be shut down (after death-timeout) because they won't be able to contact the Dask scheduler.

This should be somewhere in the doc if not there already ... in practice because a lot of people use notebooks, this may be a bit more complicated to explain for both python scripts and notebooks, not 100% sure.

[JatinGharat94]

Thankyou @lesteve for providing solutions.

I have already tried Option 1 but it consumes around 90% memory on Login Node which is highly ineffective.

Option 2 seems to be an ideal solution for my problem but our HPC does not provide direct access to compute node. I have tried directly using one compute node to run python main-dask-script.py and it produces correct output. Since I want to test scalability of my script (function) on multiple nodes, this won't give me proper scalability results.

I have also tried Option 3 by submitting PBS jobscript using qsub jobscript.sh. The details of job-script are provided below:

#PBS -N dask-python
#PBS -q research
#PBS -l walltime=20:00:00
#PBS -l place=scatter
#PBS -l select=2:ncpus=36:vntype=cray_compute
#PBS -o /$PATH_OUT$/
#PBS -V

module load cray-netcdf/4.4.1.1.3

cd /$WORK_DIR$/

time aprun -n 8  python main-dask-script.py

Once this job is submitted, it takes forever to complete the task. I think this happens because the PBS job-scheduler gets confused with dask-scheduler.

I will provide more details on the problem once I speak with the IT admins. Thank You.

[lesteve]

Since I want to test scalability of my script (function) on multiple nodes, this won't give me proper results.

I don't understand what you mean by this. Option 2 (or any other option for this matter) will allow you to use multiple nodes as long as you use cluster.scale(jobs=your_number_of_jobs).

Maybe reading this could help: https://jobqueue.dask.org/en/latest/howitworks.html.

[JatinGharat94]

I am measuring computation time, speedup and scalability of my script. So I am doing something like serial vs parallel (and in parallel, how much I can scale my application). For example, I can scale from 2 cores to 100 cores but at some point the computational time will saturate.

[lesteve]

Sorry I was not explicit enough: the part I did not understand was why Option 2 will not give you "proper results" regarding your scalability tests.

[JatinGharat94]

There is only a single node present on Login Node. So I cannot scale to multiple nodes unless my code runs on compute nodes.

[lesteve]

There seems to be some misunderstanding somewhere but I am having trouble figuring out where it is ...

The point of dask-jobqueue is to launch jobs through your job scheduler which is where the bulk of your computation happens:

• you launch main-dask-script.py from a login node (Option 1) or compute node (Option 2 or 3)

• cluster.scale will launch jobs through your job schdeduler (e.g. qsub for PBS)

• client.submit will run the Python function inside the jobs that were launched by cluster.scale

• For Dask Dataframes, Dask array, xarray (that seems to be what you are doing), once you create your Dask client, this will become the default client for your script and everything you will do with Dask dataframes, Dask array, xarray (not sure about this one) will run inside the jobs that were created by cluster.scale

• For an overview of dask-jobqueue: https://jobqueue.dask.org/en/latest/howitworks.html

• https://examples.dask.org/dataframe.html may be a good place to start for generic Dask dataframe

• for xarray, I am not sure but there are probably some Pangeo material available somewhere

[JatinGharat94]

I run my program in the following manner:

1. Login to my Login node and start jupyter-notebook
2. Using dask-jobqueue, I submit my cluster information to dask-client. client = Client(cluster)
3. Once the workers are created, I perform my task.
4. After finishing all my computations, I close my client. client.close()

This is what I am doing from a couple of days.
Now according to the system admins, I must include the aprun command which is quite similar to mpirun and mostly used in cray compute systems so my job gets submitted to compute nodes.

I also tried the other way round by submitting my main-dask-script.py along with PBS script qsub job-script but the task kept running forever.

My datasets are in .nc (netCDF4) format, so I am using xarray.

Am I following the correct steps? Thank You for your suggestions.

[ocaisa]

What you are trying to do is not really supported behaviour for dask_jobqueue (be warned I am not a maintainer). If you need to use aprun you will probably get a lot more mileage from the special Dask package that comes from Cray (see https://user.cscs.ch/computing/data_science/dask/ where the CSCS setup is described).

[lesteve]

Thanks @ocaisa for your input ! You certainly don't need to be a maintainer to have something useful to say. In this particular case you seem to have a better understanding of the situation than me ...

@JatinGharat94 I don't know anything about Cray, I don't understand why you need aprun and what the difference is vs qsub so I am a bit lost here I have to admit. If I had to guess I would hazard that your problem is a bit specific to the particular cluster you are trying to use dask-jobqueue on.

[ocaisa]

@lesteve Yes, it sounds like it is specific. If I understand the Cray systems correctly you always need to use aprun to execute something on the backend, otherwise it will still run on the frontend even if you are in the context of a job. A workaround is probably to add a kwarg when creating the cluster

python="aprun python"

so that aprun appears in the jobscript. I was reluctant to suggest that since I expect you will run into other issues.