inconsistent processing time in dask distributed fastparquet

Question

I have a parquet file in hive format and snappy compression. It fits in memory and a pandas.info provides the following data.

Number of rows per group in parquet file is just 100K

>>> df.info()
<class 'pandas.core.frame.DataFrame'>
Index: 21547746 entries, YyO+tlZtAXYXoZhNr3Vg3+dfVQvrBVGO8j1mfqe4ZHc= to oE4y2wK5E7OR8zyrCHeW02uTeI6wTwT4QTApEVBNEdM=
Data columns (total 8 columns):
payment_method_id         int16
payment_plan_days         int16
plan_list_price           int16
actual_amount_paid        int16
is_auto_renew             bool
transaction_date          datetime64[ns]
membership_expire_date    datetime64[ns]
is_cancel                 bool
dtypes: bool(2), datetime64[ns](2), int16(4)
memory usage: 698.7+ MB

Now, making some simple calculation with dask I get the following timings

Using threading

>>>time.asctime();ddf.actual_amount_paid.mean().compute();time.asctime()
'Fri Oct 13 23:44:50 2017'
141.98732048354384
'Fri Oct 13 23:44:59 2017'

Using distributed ( local cluster)

>>> c=Client()
>>> time.asctime();ddf.actual_amount_paid.mean().compute();time.asctime()
'Fri Oct 13 23:47:04 2017'
141.98732048354384
'Fri Oct 13 23:47:15 2017'
>>>

That was OK, about 9 seconds each.

Now using multiprocessing, here comes the surprise...

>>> time.asctime();ddf.actual_amount_paid.mean().compute(get=dask.multiprocessing.get);time.asctime()
'Fri Oct 13 23:50:43 2017'
141.98732048354384
'Fri Oct 13 23:57:49 2017'
>>>

I would expect multiprocessing and distributed/local cluster to be on the same order of magnitude with possibly some differences with threading ( for good or bad)

However, multiprocessing is taking 47 times more time to make a simple mean across a in16 colum?

My env is just a fresh conda install with required modules. No handpicking of anything.

why is there this differences ?? I can't manage dask/distributed to have predictable behavior as to be able to wisely choose between the different scheduler depending on the nature of my problem.

This is just a toy example but I have been unable to get an example aligned to my expectations ( as my understanding of reading the docs as least).

Is there anything I should be keeping in the back of my mind ? or am I just completely missing the point?

Thanks

JC

mdurant mdurant · Accepted Answer · 2017-10-17T19:30:55

With the threaded scheduler, each task has access to all the memory of the process - all of the data in this case - and can therefore do its calculations without any memory copying.

With the distributed scheduler, the scheduler knows which thread and which worker is producing the data required by a subsequent task, or already has that data in memory. The cleverness of the scheduler is specifically geared towards moving computation to the right worker, to avoid data communication and copying.

Conversely, the multiprocess scheduler tends to send task results to and from the main process, which can involve a lot of serialisation and copying. Some tasks can be fused together (combining tasks by calling many python functions in a chain), but some cannot. Any serialisation and copying takes CPU effort and, probably more important for you, memory space. If your original data is a significant fraction of the system total, you are probably filling up physical memory, resulting in the big factor slow-down.

inconsistent processing time in dask distributed fastparquet

1 Answers