Bandwidth during dataframe shuffles is low

[mrocklin]

I'm running the following experiment with partition frequencies of both '7d' and '30d'

from dask.distributed import Client, wait, progress
c = Client('localhost:8786')
c

import dask.dataframe as dd
df = dd.demo.make_timeseries('2000', '2001',
                             {'x': float, 'y': float, 'id': int},
                             freq='1s', partition_freq='7d', seed=2)
df = df.persist()
wait(df)

start = time()
df2 = df.set_index('id').persist()
wait(df2)
end = time()
print(end - start)  # trying to optimize this

Profiles

• Seven day (smaller) partitions
• Thirty day (larger) partitions

Analysis

Bandwidths in the small case are poor. They're close to 15MB/s per chunk. There is a lot of overlapping though. We tend to get somewhere around 50MB/s and 70MB/s. (This is for the whole machine). In the larger case they're better, around 120MB/s on average peaking up to 250MB/s. This is all on localhost though, so we should expect better.

The time to instantiate a new connection can takes a surprising amount of time (100ms or so) with wide spread, which might contribute a bit to the poor bandwidths. This should be at or below a millisecond.

My first guess on all of this is that our network is less responsive due to GIL-holding computations in Pandas (cc @jreback) but this is just a guess.

Some things we can do

1. Reduce serialization cost of pandas dataframes (see Pandas serialization #931 and https://issues.apache.org/jira/browse/ARROW-376) (cc @wesm)
2. Pool inter-worker connections (will try this next)
3. Investigate gil-free pandas operations (these computations get poor speedups in a thread pool, so this is presumably still a live issue (cc @jreback))
4. Byte-copy-free tornado IOStreams? (cc @pitrou)

The odds of any individual issue here resolving the problem is low so individually these are all probably pretty low priority.

[jreback]

what kinds of ops are dd.set_index(...) actually doing?

[mrocklin]

what kinds of ops are dd.set_index(...) actually doing?

Looking at the profiles may give some understanding here. In summary:

1. Splitting dataframes into groups

   def shuffle_group(df, col, stage, k, npartitions):
       if col == '_partitions':
           ind = df[col].values % npartitions
       else:
           ind = partitioning_index(df[col], npartitions)
       c = ind // k ** stage % k
       g = df.groupby(c)
       return {i: g.get_group(i) if i in g.groups else df.head(0) for i in range(k)}

2.  Splitting those dicts into different components (just getitem)
3.  Concatenation along rows
4.  Sorting along the index (towards the end)

Step 1 is the only thing that happens during communication heavy periods and lasts for any real duration.

[jreback]

(this is on a local machine). looks like most of the bottlenecks are actually the shuffle (green) or am I reading this wrong?

[jreback]

[mrocklin]

I recommend starting a dask-worker with --nprocs 8 --nthreads 1 to force inter-process communication.

[mrocklin]

A couple other anecdata points:

1. Replacing a numeric column with a text column doubles or triples the shuffle duration. To be clear we're not setting this text column as the index, it's just along for the ride.
2. Sorting a Spark DataFrame with the same data is around 30% faster in the numeric case and a few times faster in the text case (at least with my current configuration).

[jreback]

[jreback]

yep a bit different :>

[mrocklin]

You may also enjoy looking at the following diagnostic pages:

• http://localhost:8789/system
• http://localhost:8789/main.html
• http://localhost:8789/crossfilter

These give a good view of what a particular worker is doing

[jreback]

sorting (a single column) is pretty much using np.argsort which is gil releasing

[mrocklin]

Any thoughts on how to improve the shuffle_group function? The goal here is to hash a particular column and then separate it into different sub-dataframe components based on those hash values.

It might be simplified as follows

def shuffle_group(df, col, stage, k, npartitions):
    ind = hash_pandas_object(df[col]) % npartitions
    c = ind // k ** stage % k
    g = df.groupby(c)
    return {i: g.get_group(i) if i in g.groups else df.head(0) for i in range(k)}

[jreback]

if this is my scheduler info, the workers how do I look at the worker?

{'address': 'tcp://127.0.0.1:8786',
 'id': '98dfdc62-0023-11e7-a34b-6c4008997976',
 'services': {'http': 53003},
 'type': 'Scheduler',
 'workers': {'tcp://127.0.0.1:52979': {'host': '127.0.0.1',
   'last-seen': 1488554224.744344,
   'last-task': 1488553901.274083,
   'local_directory': '/var/folders/h3/mr_r3bkj5yg0pbx9fr3tk1r00000gp/T/worker-tbgc3emo',
   'memory_limit': 1288490188.0,
   'name': 'tcp://127.0.0.1:52979',
   'ncores': 1,
   'services': {},
   'time-delay': 0.006539106369018555},

[mrocklin]

If you have bokeh installed and start the worker with dask-worker ... it should come up automatically. You would have to be using a version of dask.distributed released sometime this year.

[mrocklin]

The worker bokeh servers aren't deployed if you're using the Python LocalClient or Client() convenience setup solutions.

[jreback]

ahh, ok

shuffle_group(df, 'A', 10, 10, 10)

what kind of parameters are you calling this with? whats the dtype (and len) of the column? (IOW is the the actual column from a pandas DataFrame (not dask) above?)