Local distributed scheduler way slower than local processes scheduler

[aberres]

I am right now investigating the overhead of dask to get a feeling if dask might make sense for my workload (medium-sized data frames with short operations).

I came up with an (dumb) example where scheduler='processes' outperformed scheduler='single-threaded' as expected.
I am constructing a pipeline where I do nothing more but routing a dataframe with minimal manipulations. As the pipeline has no dependencies I would expect no worker-worker communication. All data could be available locally on the worker
The larger the dataframes are, the slower the process scheduler is though - as expected.

As next step I was trying the very same example with the local distributed scheduler. After all the documentation tells me

the distributed scheduler described a couple sections below is often a better choice today.

The results were not very pleasing. Compared to the process scheduler the distributed scheduler is up to 10x slower.
Is this expected and the distributed scheduler does simply not make sense in this case?
Am I doing something wrong?
Without knowing the internals of the scheduler I did expect it to perform relatively well (and comparable to the process scheduler) in the given case as data (in my understanding) never needs to leave a worker.

import dask
import time
import pandas as pd
import numpy as np
from distributed import Client

def process1(data):
    return data

def process2(inp):
    time.sleep(0.001)
    return inp + 1

def process3(inp):
    return inp

tasks = []
for i in range(20):
    data = pd.DataFrame(np.zeros((1000, 100)))
    processed = dask.delayed(process1)(data)
    for _ in range(100):
        processed = dask.delayed(process2)(processed)
    
    processed2 = dask.delayed(process3)(processed)
    
    tasks.append(processed2)

%time r0 = dask.compute(tasks, scheduler='processes')
# Wall time: 1.67 s

%time r1 = dask.compute(tasks, scheduler='single-threaded')
# Wall time: 4.57 s

client = Client()
%time r2 = dask.compute(tasks)
# Wall time: 14.4 s

[mrocklin]

I recommend taking a look at the dashboard during computation. Does anything pop out?

…

On Wed, Oct 31, 2018 at 10:47 AM Armin Berres ***@***.***> wrote: I am right now investigating the overhead of dask to get a feeling if dask might make sense for my workload (medium-sized data frames with short operations). I came up with an (dumb) example where scheduler='processes' outperformed scheduler='single-threaded' as expected. I am constructing a pipeline where I do nothing more but routing a dataframe with minimal manipulations. As the pipeline has no dependencies I would expect no worker-worker communication. All data could be available locally on the worker The larger the dataframes are, the slower the process scheduler is though - as expected. As next step I was trying the very same example with the local distributed scheduler. After all the documentation tells me the distributed scheduler described a couple sections below is often a better choice today. The results were not very pleasing. Compared to the process scheduler the distributed scheduler is up to 10x slower. Is this expected and the distributed scheduler does simply not make sense in this case? Am I doing something wrong? Without knowing the internals of the scheduler I did expect it to perform relatively well (and comparable to the process scheduler) in the given case as data (in my understanding) never needs to leave a worker. import daskimport timeimport pandas as pdimport numpy as npfrom distributed import Client def process1(data): return data def process2(inp): time.sleep(0.001) return inp + 1 def process3(inp): return inp tasks = []for i in range(20): data = pd.DataFrame(np.zeros((1000, 100))) processed = dask.delayed(process1)(data) for _ in range(100): processed = dask.delayed(process2)(processed) processed2 = dask.delayed(process3)(processed) tasks.append(processed2) %time r0 = dask.compute(tasks, scheduler='processes')# Wall time: 1.67 s %time r1 = dask.compute(tasks, scheduler='single-threaded')# Wall time: 4.57 s client = Client()%time r2 = dask.compute(tasks)# Wall time: 14.4 s — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#2327>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszEeZVD8XSCfce7ldhLChoQiijn_rks5uqbfugaJpZM4YEPTA> .

[aberres]

Well, not sure.
There is a lot of white. The greenish bars are the process2 calls as expected.

[mrocklin]

It's an interesting problem. It looks like there is a 5-10ms gap between tasks here. I tried a few things:

1. Switch to tornado master, which reduces scheduler overhead a bit: no real difference
2. Look at the /profile-server route of the diagnostic pages to see what's taking up time on the scheduler. It looks like it's not terribly burdened during this time, so it's probably not the scheduler slowing down too much
3. Increase the size of tasks and the number of concurrent tasks and things look quite a bit better, but that's hiding the root problem
4. I took a look a the worker diagnostic pages (click on one of the bars in the memory or processing plots to get access) and noticed that they seem to think that they're continuing to work, assuming that you increase the number of concurrent tasks (the 20 number you have)

So, no strong conclusions. I'm surprised to see a latency of 10ms back and forth from the worker to scheduler.

[martindurant]

Ran it for fun, and found that the workers spend a surprising amount of time in distributed.sizeof.safe_sizeof, which takes about 0.5ms for the data (which is the input and output of every task).

In [19]: %timeit data.values.nbytes + data.index.nbytes + data.columns.nbytes
5.45 µs ± 37.9 ns per loop (mean ± std. dev. of 7 runs, 100000 loops each)
800160

In [20]: %timeit distributed.sizeof.sizeof(data)
456 µs ± 4.06 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)
802080

[aberres]

Another observation: When massively increasing the data frames (10000, 1000) the distributed scheduler seems to perform better in comparison.

Still, a single thread is way faster. I am wondering one thing: Isn't this a problem which should be easy to parallelize as all data should be available on each worker once the first data frame is passed? Or is even inside a worker the bookkeeping, when passing data from function to function, simply relatively complex compared to calling native functions?

[mrocklin]

The scheduler keeps track of dependencies, so every time a worker finishes a task it checks in with the scheduler, and the scheduler tells it the next task it should work on. As you increase the `20` number above you'll get more concurrency, which should help to reduce worker idling. However as you increase the `100` number you're still sequential, which will require checking in with the scheduler.

…

On Thu, Nov 1, 2018 at 4:29 AM Armin Berres ***@***.***> wrote: Another observation: When massively increasing the data frames (10000, 1000) the distributed scheduler seems to perform better in comparison. Still, a single thread is way faster. I am wondering one thing: Isn't this a problem which should be easy to parallelize as all data should be available on each worker once the first data frame is passed? Or is even inside a worker the bookkeeping, when passing data from function to function, simply relatively complex compared to calling native functions? — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#2327 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AASszEYXkQE3iDp-Iy36lGWeIVsNr9j-ks5uqrDSgaJpZM4YEPTA> .

[aberres]

Makes sense.

But this overhead should be constant and not change with the size of the data frames passed between the tasks, should it? Given that the data is processed on the same worker.

[mrocklin]

That's correct. But having larger dataframes means that there is probably more work to do to cover the overhead when contacting the scheduler.