Several millions tasks and Client.map

[guillaumeeb]

Hi,

I'm facing an issue that has already been raised in other more complicated forms, see pangeo-data/pangeo/issues/266, dask/dask/issues/3514, and probabaly other places. I've read those threads, but I cannot see yet a simple solution or good workaround for my problem.

I'm using Dask to do some really basic embarrassingly parallel problem:

• I've got a scientific code, that I will call using subprocess
• I need to run int with exactly 3123120 different set of parameters.

So here is basically what I do:

cmd_lines = []
for param1 in param1_set
    for param2 in param2_set
         # Some other nested loops
         cmd_lines += [' '.join(("myscript.sh", param1, param2, ... ))]

def run_process(cmd):
    os.chdir(WORK_DIR)
    process = Popen(cmd.split(' '), stdout=PIPE)
    (output, err) = process.communicate()
    exit_code = process.wait()
    return output, err

futures = client.map(run_process, cmd_lines)
results = client.gather(futures)

When taking only the first 100,000 elements of cmd_lines, this works with some minor problems:

• It takes between 40 seconds and 1m30s before my workers do anything, I imagine this is the time of building the graph and sending to worker.
• Gather takes about 20 seconds, this seems better than the scattering though.

When trying with the full 3 billions parameters, I can't see any computation starting, and the Dashboard is almost not responding anymore.

So an easy workaround is just going 100,000 tasks by 100,000, but I'm not satisfied by this, and it unnecessarily complexifies the really simple code here.

To sum up, I'd like to have a solution for two problems:

• First one is more of an annoyance: the time taken to actually start computation, without any information on the Dashboard. For such an embarrassingly parallel workload, I would even expect the tasks to start right after submission, they could be sent to workers in a streaming way, and we should not wait for 100,000 of tasks to be all sent to workers to begin.
• Second one which is probably very related, scheduler should handle several billion tasks at a time.

[mrocklin]

First one is more of an annoyance: the time taken to actually start computation, without any information on the Dashboard. For such an embarrassingly parallel workload, I would even expect the tasks to start right after submission, they could be sent to workers in a streaming way, and we should not wait for 100,000 of tasks to be all sent to workers to begin.

If Dask were just about single tasks then this would be doable. However for full task graphs we would first need to find nice partitions of the graph so that we could send things up piece by piece. I think that what is needed here is an algorithm to partition a task graph very very quickly in the common case. If we have this then we can send up large graphs piece by piece. This is something that someone could research without being deeply knowledgable about the internals of the scheduler.

So, question, is there an algorithm to partition billion node graphs quickly? This would obviously need to operate in linear time, and doesn't need to be perfect, just do some good.

Second one which is probably very related, scheduler should handle several billion tasks at a time.

This fights against per-task overhead which many groups are currently trying to add to in order to get new features. Balancing between many-cheap-tasks and few-expensive-tasks is a social as well as a technical challenge.

In your siutation I typically recommend dask.bag. See http://dask.pydata.org/en/latest/delayed-best-practices.html#avoid-too-many-tasks

[guillaumeeb]

Thanks @mrocklin, current solution was batching, I will have a look at bags. Hope it wont look to Sparkish, I was trying to avoid RDD for that kind of multi tasking.

So, question, is there an algorithm to partition billion node graphs quickly?

If I correctly understand what you say, it is quite simple here, given I am using client.map(). For me the algorithm is just taking a given batch size and split the input collection or iterator. Is there something I don't get?

[mrocklin]

If I correctly understand what you say, it is quite simple here, given I am using client.map(). For me the algorithm is just taking a given batch size and split the input collection or iterator. Is there something I don't get?

Ideally we would find a more general way to handle this at a bit lower level that could more parts of Dask than just the map method. For example if someone has your same workflow, but they use dask.delayed, is there a way that we can help them as well?

[guillaumeeb]

Could you point to some places to look at in the codebase for where the graph partitioning occurs?

[mrocklin]

No graph partitioning occurs currently, which is why no work happens until all of your computation is sent up to the scheduler.

…

On Mon, Aug 13, 2018 at 4:26 PM, Guillaume EB ***@***.***> wrote: Could you point to some places to look at in the codebase for where the graph partitioning occurs? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#2181 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszPsfbAp3cTtkzUaPNJH_IYBO7077ks5uQeD9gaJpZM4V6DDb> .

[guillaumeeb]

Okay, this makes sense...

So just one more thing, why is it taking so long to send the computation to the scheduler in that case? It is the 1ms per task thing? So 100 seconds for 100,000 tasks?

[mrocklin]

Something like that yes. This could be worse depending on if you are sending data, or if your function is large when serialized, etc.. 1ms is a bit conservative in best case. I think that something like 200us is doable today if you're careful.

…

On Mon, Aug 13, 2018 at 4:49 PM, Guillaume EB ***@***.***> wrote: Okay, this makes sense... So just one more thing, why is it taking so long to send the computation to the scheduler in that case? It is the 1ms per task thing? So 100 seconds for 100,000 tasks? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#2181 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszN98bIfG10HmUuq4u2Uy1CUhBALlks5uQeZngaJpZM4V6DDb> .

[mrocklin]

We could also look into special casing map if there is some simple way to do it. Perhaps we look at the lengths of the iterators and if they're very long then we cut things up and call map many times. We would have to be careful about priorities. I wouldn't be surprised to find a couple other interesting challenges along the way too. On Mon, Aug 13, 2018 at 4:59 PM, Matthew Rocklin <mrocklin@anaconda.com> wrote:

…

Something like that yes. This could be worse depending on if you are sending data, or if your function is large when serialized, etc.. 1ms is a bit conservative in best case. I think that something like 200us is doable today if you're careful. On Mon, Aug 13, 2018 at 4:49 PM, Guillaume EB ***@***.***> wrote: > Okay, this makes sense... > > So just one more thing, why is it taking so long to send the computation > to the scheduler in that case? It is the 1ms per task thing? So 100 seconds > for 100,000 tasks? > > — > You are receiving this because you were mentioned. > Reply to this email directly, view it on GitHub > <#2181 (comment)>, > or mute the thread > <https://github.com/notifications/unsubscribe-auth/AASszN98bIfG10HmUuq4u2Uy1CUhBALlks5uQeZngaJpZM4V6DDb> > . >

[guillaumeeb]

1ms is a bit conservative in best case. I think that something like 200us is doable today if you're careful.

Actually, I've more of a 45 to 60 seconds overhead, so 500us per task.

For the rest, I'd be happy to have a look at all that, however I expect it to be quite difficult, and I'm already engaged in some other work on distributed or dask-jobqueue, so probably not immediately.

[guillaumeeb]

In your situation I typically recommend dask.bag. See http://dask.pydata.org/en/latest/delayed-best-practices.html#avoid-too-many-tasks

Dask bag works great for batching small calls. Code is not more complicated (except I want to specify the number of records per partitions, and not the number of partitions):

#I want 100 records by partition
npartitions = len(cmd_lines) // 100 + 1
b = db.from_sequence(cmd_lines, npartitions=npartitions)
result = b.map(run_process).compute()

Just a little sad when looking at dashboard to see task advancing with big jumps every minutes or so. It was prettier when all was moving continously 🙂.

When launching on 3 millions records, with 30 000 partitions, I still have an delay of nearly two minutes before the computation begins. But after that it's perfect.

[guillaumeeb]

Were should I look if I wanted to reduce the few minutes overhead of submitting a high number of tasks?

And second point, would it be interesting to have the kwarg partition_size as a pendant to npartitions in from_sequence?

[guillaumeeb]

Were should I look if I wanted to reduce the few minutes overhead of submitting a high number of tasks?

I've should had put that better: where should I look if I wanted to see if some optimization is possible to reduce the overhead? Trying something like you mentioned above for example:

Perhaps we look at the lengths of the iterators and if they're very long then we cut things up and call map many times. We would have to be careful about priorities

[mrocklin]

This would be probably at the very top of map

def map(self, func, seq, ..., partition_size=None)
    if partition_size and len(seq) > partition_size:
        return sum([self.map(func, batch, ...) for batch in toolz.partition_all(seq, partition_size)], [])
    ...

[jurajmichalak1]

I read this thread and I'm very confused.
I see that in @guillaumeeb 's source code, there are no dependencies between tasks he is going to submit to cluster. So there is no need to do any complicated task graph computations on scheduler, before starting executing those tasks on workers. Could you possibly @mrocklin explain this?

First one is more of an annoyance: the time taken to actually start computation, without any information on the Dashboard. For such an embarrassingly parallel workload, I would even expect the tasks to start right after submission, they could be sent to workers in a streaming way, and we should not wait for 100,000 of tasks to be all sent to workers to begin.

I think you are exactly right, I was expecting that it will start streaming those tasks to workers, processing them and receiving results. The unfortunate thing is that there is no warning in Dask documentation about how Client.map() works. To me it seems that it tries to serialize whole iterator / generator to the workers at once. I had same issue that I used Client.map() on huge array of images and my workers started to run out of memory.

I think that this is related to my problem described here:
#864 (comment)
The solution is to use Client.submit() and control amount of submitted and unprocessed tasks at some "maxsize" value.

[muammar]

I also get very confused when reading this thread. For embarrassingly parallel computations what should we be using?

1. client.submit as suggested by @jurajmichalak1.
2. dask.bag (and I wonder if when using dask_jobqueue with a dask distributed Client I will get futures or concrete values).
3. client.map?

It seems to me that using client.submit might work better because one avoids giving too much workload to the scheduler. However, this is more from the point of view of what a user would expect (I will elaborate). When using client.submit you will automatically see in the dashboard that something is happening unlike client.map, dask.bag, client.compute and client.persist that rely on the scheduler and make the delay in the dashboard considerable. Any input would be greatly appreciated. Is this the type of thing that whatever it works better is the way to go?