Add support for job arrays

[sntgluca]

Dear dask community,

I am moving my first steps with Dask and distributed, and I wish to bring up a couple of ideas and suggestions.
The first one is about enhancing the integration with the schedulers.

Dask workers are usually requested in bulk.
When the schedulers allow it, i.e. with SGE, I think it's a good practice to submit those as job (task) arrays rather than multiple independent jobs.

In systems with heavy loads it would make the work of the scheduler much easier.

I don't know the code well enough yet to come up with a complete solution, I think there are multiple places where the code needs changes. So I would like to be sure it's a good idea before working on it.

What do you think?

[guillaumeeb]

The first thought that comes to my mind here is that we've designed dask-jobqueue with simplicity in mind, as discussed and argumented by @jhamman here: #84 (comment).

The discussion was about multi nodes reservation at first, but I feel this apply, even with a lesser extend, to job arrays. I'd be curious to have @jhamman advice here.

What goes into your direction @sntgluca is that I believe that job arrays have an implementation almost every job queue softwares. But still, I'm not sure if this is an easy thing to implement in dask-jobqueue, and we certainly don't want to make it too complex.

One question if we implemented this, how will we able to easily implement scale_up and scale_down functions with arrays? Is it always easy to qdel or equivalent only part of an array of jobs? Will this be compatible with the internal state (running_jobs, pending_jobs ...) we store?
I think if we wanted to use arrays, we should do it systematically to keep only simple job templates...

Looks like a quite heavy change.

[jhamman]

@sntgluca - in the early days of dask on hpc, we explored using job arrays as a tool for deploying dask (dask/distributed#1260 (comment)). IIRC, we found this approach was generally difficult to abstract to generic array jobs. It might we worth reading https://docs.dask.org/en/latest/setup/hpc.html if you have not already.

I think there is probably a use case for array jobs with dask but I don't think dask-jobqueue is the right place. While I do see how they may lighten the load on the (SGE) scheduler, they are far less capable of dynamic scaling than what we have now.

[mrocklin]

Some IT groups prefer job arrays over many single jobs because it reduces stress of the resource manager (SLURM, SGE, etc..) They view submitting thousands of independent jobs as an anti-pattern. If there is an easy way to achieve this while still maintaining the current design then I think that it would be welcome. The challenge is to find something that works out well in both cases. Sometimes with job arrays (my only experience here is with SGE) it just produces many new Job ID like 10.1, 10.2, 10.3, 10.4 and those jobs proceed as though we had submitted them independently. You can still manage them independently. In this situation we might consider submitting an array job whenever we scale up by a large number at once. However, I don't know if my understanding here is correct, and even if so I don't know if this would generalize. I think that the general answer here is that "if this can be made to work without adding much complexity, and if it demonstrably improves the experience for some users or administrators then we'd be happy to have it". Proposing a design and implementing a proof of concept that acheives this would be a good activity for anyone who wants to push forward on this issue.

…

On Thu, Nov 15, 2018 at 5:15 PM Joe Hamman ***@***.***> wrote: @sntgluca <https://github.com/sntgluca> - in the early days of dask on hpc, we explored using job arrays as a tool for deploying dask (dask/distributed#1260 (comment) <dask/distributed#1260 (comment)>). IIRC, we found this approach was generally difficult to abstract to generic array jobs. It might we worth reading https://docs.dask.org/en/latest/setup/hpc.html if you have not already. I think there is probably a use case for array jobs with dask but I don't think dask-jobqueue is the right place. While I do see how they may lighten the load on the (SGE) scheduler, they are far less capable of dynamic scaling than what we have now. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#196 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszCA0tLEMSsdgzJNu-FgUgYZUC3xTks5uved6gaJpZM4Yffg1> .

[belledon]

Not sure if this will help as I am relatively new to using dask but I have played around with using job arrays. Right now you can force

        n = 100
        params = {
            'cores' : 2,
            'memory' : '1000MB',
            'processes' : 1,
            'walltime' : '180',
            'job_extra' : [
                '--array 0-{0:d}'.format(n - 1), # submit array of sizen
                '--requeue',
            ],
            'env_extra' : [
                'JOB_ID=${SLURM_ARRAY_JOB_ID%;*}_${SLURM_ARRAY_TASK_ID%;*}',
            ],
        }

which will result with each worker having a unique id by overwriting JOB_ID.

#!/usr/bin/env bash

#SBATCH -n 1
#SBATCH --cpus-per-task=2
#SBATCH --mem=954M
#SBATCH -t 180
#SBATCH --array 0-99
#SBATCH --requeue
JOB_ID=${SLURM_JOB_ID%;*}

JOB_ID=${SLURM_ARRAY_JOB_ID%;*}_${SLURM_ARRAY_TASK_ID%;*}

This works just fine when using simple setups since each worker gets added to the pool even if the scheduler complains that it is unknown. However, each call the scale will result in n new workers, so client.scale(i) means n * i workers.

[guillaumeeb]

Some IT groups prefer job arrays over many single jobs because it reduces stress of the resource manager (SLURM, SGE, etc..) They view submitting thousands of independent jobs as an anti-pattern.

Yep, this is clearly the case at CNES, but it is when one doesn't use Dask :). We've a limitation of 200 concurrent individual jobs, or 10 job arrays, but with up to 100,000 jobs each. But 200 dask-jobqueue jobs with 24 cores is already 4800 cores, that's 2/3 of our cluster... More importantly, in my point of view, dask-jobqueue can play a role similar to job-arrays, as once you've booked a number of resources, you can start computing a lot of individual tasks, and you also don't have to wait for all resources to come up.

However, I don't know if my understanding here is correct, and even if so I don't know if this would generalize.

I think your understanding is correct, but as you say we first need here to check if this generalize on all job queue system. It seems to be the case for PBS, Slurm and SGE at least.

I think that the general answer here is that "if this can be made to work without adding much complexity, and if it demonstrably improves the experience for some users or administrators then we'd be happy to have it".

I'm perfectly fine with that, though I don't know if it is worth the effort.

[mrocklin]

though I don't know if it is worth the effort

I agree. For me personally it's not worth the effort. It sounds like it's the same for you. If this is very important for someone else though then I want them to know that they are welcome to do it.

…

On Mon, Nov 19, 2018 at 3:48 PM Guillaume Eynard-Bontemps < ***@***.***> wrote: Some IT groups prefer job arrays over many single jobs because it reduces stress of the resource manager (SLURM, SGE, etc..) They view submitting thousands of independent jobs as an anti-pattern. Yep, this is clearly the case at CNES, but it is when one doesn't use Dask :). We've a limitation of 200 concurrent individual jobs, or 10 job arrays, but with up to 100,000 jobs each. But 200 dask-jobqueue jobs with 24 cores is already 4800 cores, that's 2/3 of our cluster... More importantly, in my point of view, dask-jobqueue can play a role similar to job-arrays, as once you've booked a number of resources, you can start computing a lot of individual tasks, and you also don't have to wait for all resources to come up. However, I don't know if my understanding here is correct, and even if so I don't know if this would generalize. I think your understanding is correct, but as you say we first need here to check if this generalize on all job queue system. It seems to be the case for PBS, Slurm and SGE at least. I think that the general answer here is that "if this can be made to work without adding much complexity, and if it demonstrably improves the experience for some users or administrators then we'd be happy to have it". I'm perfectly fine with that, though I don't know if it is worth the effort. — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#196 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszEXz68rxVE5mOeudQPlb-4ijSYxqks5uwxkhgaJpZM4Yffg1> .

[sntgluca]

Hi all,

thanks for the all the follow up and sorry for delaying my answer. I personally find it an important asset. I speak from my personal experience as a HPC user working in close contact with my system administrators.

One thing to keep in mind is that dask is pretty new, and I am not sure how many syadmin are actually following your progress :) So it's not easy to measure the real impact there.

one question if we implemented this, how will we able to easily implement scale_up and scale_down functions with arrays? Is it always easy to qdel or equivalent only part of an array of jobs? Will this be compatible with the internal state (running_jobs, pending_jobs ...) we store?
I think if we wanted to use arrays, we should do it systematically to keep only simple job templates...

I am positive this can be done in scale_up and scale_down. But there are other parts of the code requiring attention. For instance, a task array is returned as a unique job, not the whole list of jobs, messing up with the pending_jobs list.

Before you decide to close this, is it ok if I give it a try? What would you expect a newcomer do ;)?

[mrocklin]

Even though it's a single job (like 123) the individual components of that job are sometimes addressable (like 123.456 for the 456th component of job 123).

Before you decide to close this, is it ok if I give it a try?

Definitely

What would you expect a newcomer do ;)?

Presumably you would modify the code to achieve the results that you want. You would also provide tests that clearly communicate your desired change, and also demonstrate that that change has occurred. You should also probably read through the general Dask developer guidelines here: http://docs.dask.org/en/latest/develop.html#development-environment

…

On Tue, Nov 27, 2018 at 8:17 AM sntgluca ***@***.***> wrote: Hi all, thanks for the all the follow up and sorry for delaying my answer. I personally find it an important asset. I speak from my personal experience as a HPC user working in close contact with my system administrators. One thing to keep in mind is that dask is pretty new, and I am not sure how many syadmin are actually following your progress :) So it's not easy to measure the real impact there. one question if we implemented this, how will we able to easily implement scale_up and scale_down functions with arrays? Is it always easy to qdel or equivalent only part of an array of jobs? Will this be compatible with the internal state (running_jobs, pending_jobs ...) we store? I think if we wanted to use arrays, we should do it systematically to keep only simple job templates... I am positive this can be done in scale_up and scale_down. But there are other parts of the code requiring attention. For instance, a task array is returned as a unique job, not the whole list of jobs, messing up with the pending_jobs list. Before you decide to close this, is it ok if I give it a try? What would you expect a newcomer do ;)? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#196 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszJ29-QASVNYSjE2PLO0LBSvsoTzXks5uzTuCgaJpZM4Yffg1> .

[sntgluca]

Thanks for the guidelines @mrocklin,
will follow up as soon as I have something to say!

[sntgluca]

Hi, I have a working proof of concept to use job arrays for the SGECluster class.
I can achieve it by rewriting the start_cluster() method and modifying the self._command_template attribute.

Before moving further, I have a design question.

How do you think the change shall be hooked into the current code? I suggest the use of job arrays be optional, for instance activated by a boolean

dask_jobqueue.SGECluster(use_task_array=True)

This idea seems consistent with the rest of the code and shall be robust, but might make the design a bit dull: if use_task_array do this else do that...

[guillaumeeb]

This idea seems consistent with the rest of the code and shall be robust, but might make the design a bit dull: if use_task_array do this else do that...

I'm a bit afraid of this yes... But defaulting to job arrays sounds bad too. Is there a lot of places where the if statement would be used? Can we limit this? If it is only in start_cluster() or above, this should be okay.

So I'm slightly in favor of using the boolean activator you propose.

[lesteve]

I commented on the the #217 but I'll comment on the issue too with more complete thoughts. My feeling echoes the feeling of others in this issue: task array support is not worth the effort at this point of time in dask-jobqueue.

IMO task array support is a "nice to have" feature but I am not convinced it is one of the main bottleneck for dask-jobqueue users in practice. I may be wrong about this and I would be happy to get more details about your particular use case if you can not use dask-jobqueue because it does not support task array. If we get many reports like this, I would be very happy to reprioritise task array support.

My experience with open-source projects is the following: there is quite a long (and ever growing) list of "nice to have" features but not that much time and energy on people's hands to actually implement them, review PRs, test them thoroughly, etc ... deciding what to focus on is key.

[halsafar]

Coming from a compute center with 30,000 SGE nodes. Job arrays are a must. The head node cannot handle the load without job arrays. In one of our software products we were debating a migration to Dask (from our in-house solution) and this is the only feature missing but it is enough to prevent our migration.

[lesteve]

Full disclosure: I am not very familiar with big clusters. 30_000 nodes seems like a lot though, is it number of nodes or number of cores? I know dask-jobqueue is used at Cheyenne which is ~40 in the TOP 500 and has roughly ~4000 nodes according to this

Other than this:

• what is an acceptable number of jobs to launch on your cluster with qsub, in other words what is the threshold of job numbers where you have to use job arrays?
• what is the typical size of your job arrays?
• what is the typical time a single job takes?

Something that may be worth noting: dask can be more flexible than job arrays so this may allow you to do more with the limited amount of qsub you are allowed to have (as hinted in #196 (comment)). Typically there is not a 1-to-1 mapping between "chunk or work" and job and you can use many "chunks of work" inside the same job.

Here is a snippet to demonstrate what I mean:

from dask_jobqueue import SGECluster
cluster = SGECluster(cores=1, memory='1GB')
# this will launch 4 jobs
cluster.scale(cores=4)

from dask.distributed import Client
client = Client(cluster)

def my_function(i):
    return i

argument_list = range(1000)

# the number of times you call my_function is not related to the number of jobs
futures = [client.submit(my_function, arg) for arg in argument_list]