在dask分佈式fastparquet處理時間不一致

Question

我有一個hive格式和快速壓縮的parquet文件。它適合內存，pandas.info提供以下數據。

在拼花文件每組的行數僅僅是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 

現在，做一些簡單的計算與DASK我得到以下計時

使用線程

>>>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' 

使用分佈式（本地簇）

>>> 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' 
>>> 

這很好，每個約9秒。

現在使用多，來這裏的驚喜...

>>> 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' 
>>> 

我希望多和分佈式/本地集羣是相同數量級的有可能與線程一些差異上（或好或壞）

但是，多處理多花了47倍的時間來在in16柱上做出簡單的平均值？

我的環境只是一個新鮮的conda安裝與所需的模塊。沒有任何東西的手藝。

爲什麼會有這種差異？我無法管理dask/distributed以具有可預測的行爲，以便能夠根據我的問題的性質在不同的調度程序之間進行明智的選擇。

這只是一個玩具的例子，但我一直無法得到一個與我的期望相一致的例子（正如我對閱讀文檔的理解一樣）。

有什麼我應該留在我的腦海裏？還是我完全錯過了這一點？

感謝

JC

Answer 1

與螺紋調度，每個任務都有訪問過程的所有記憶 - 所有在此情況下，數據的 - 因此可以做它的計算沒有任何記憶複製。

對於分佈式調度程序，調度程序知道哪個線程和哪個工作程序正在生成後續任務所需的數據，或者已經在內存中擁有該數據。調度程序的聰明性專門用於將計算移動到正確的工作人員，以避免數據通信和複製。

相反，多進程調度程序傾向於將任務結果發送到主進程或從主進程發送任務結果，這可能涉及很多序列化和複製。有些任務可以融合在一起（通過在鏈中調用多個python函數來組合任務），但有些不能。任何序列化和複製都需要CPU的努力，對你來說可能更重要的是內存空間。如果您的原始數據佔系統總量的很大一部分，那麼您可能會填滿物理內存，導致大的因素減慢。