1. 程式人生 > >[譯] 在 Python 中,如何運用 Dask 資料進行並行資料分析

[譯] 在 Python 中,如何運用 Dask 資料進行並行資料分析

多維度思維。來源:Pixabay

有時你通過 Python’s Pandas 開啟一個大的資料集,然後試著去獲取一些度量標準,但這時整個過程可能會突然停止。 如果你使用 Pandas 處理大資料,可能一個簡單的序列平均值都需要你等待一分鐘,我們甚至不會去呼叫 apply

。這還只是百萬級別的行數!當你的資料達到億級別時,你最好使用 Spark 或者其他方式。

我在不久之前發現了這個工具:不需要更好的基礎架構或轉換語言,就可以在 Python 中加速資料分析的方法。但是如果資料集太大,它的最終優化結果會一定的限制,但是它仍然比常規的 Pandas 擴充套件性好,可能也更符合你的問題場景 —— 尤其是不進行大量的重寫索引時。

什麼是 Dask?

Dask 是一個開源專案,為你提供 NumPy 陣列、Pandas Dataframes 以及常規 list 的抽象,允許你使用多核處理器並行執行它們的操作。

以下是來自本教程的摘錄:

Dask 提供了更高級別的 Array、Bag、和 DataFrame 集合,它們模仿 NumPy、list 和 Pandas,但允許在不適合主記憶體的資料集上並行操作。對於大型資料集,Dask 的高階集合可以取代 NumPy 和 Pandas。

這聽上去很好!為了這篇文章,我特意試用了 Dask Dataframs,並在其上運行了幾個基準測試。

閱讀文件

我首先閱讀了官方文件,看看在 Dask 的文件中的精確推薦,而不是常規 Dataframse。以下是官方文件的部分內容:

  • 操作大型資料集,即使這些資料不適用於記憶體
  • 使用盡可能多的核心來加速長時間計算
  • 在大的資料集上,通過標準的 Pandas 操作,如叢集、連線、還有時間序列計算,來對計算做分散式處理。

接下來,它列出了一些很快的場景,但前提是你在使用 Dask 資料:

  • 算術運算(對序列進行乘或加)
  • 常用聚合(均值、最小值、最大值、和等)
  • 呼叫 apply(只要它是索引,而非 groupby(‘y’),其中 y 並非索引)
  • 呼叫 value_counts()、drop_duplicates() 或 corr()
  • Locisin 和逐行選擇進行過濾

如果發現它有用,只對資料過濾進行一次小瀏覽就行。

#通過引用僅,返回 x >5 的行(根據起初的 df 寫入更改)
df2 = df.loc[df['x'] > 5]
#通過引用,僅返回x 為 0、1、2、3 或 4 的行
df3 = df.x.isin(range(4))
#通過只讀引用,僅返回 x > 5 的行(不能被寫)
df4 = df[df['x']>5]
複製程式碼

如何使用 Dask Dataframes

Dask Dataframes 具有與 Pandas Dataframes 相似的 API,只有聚合和 apply 是延遲計算,你需要通過呼叫 compute 方法來計算。為了生成一個 Dask Dataframe,你可以像在 Pandas 中那樣簡單呼叫 read_csv 方法,或只調用給定的一個 Pandas Dataframe df

dd = ddf.from_pandas(df, npartitions=N)
複製程式碼

ddf 是你使用 DASK Dataframes 匯入的名稱,而 nparitions 是一個引數,它告訴 Dataframe 你期望如何對它進行分割槽。

StackOverflow,建議將 Dataframe 劃分到你計算機核心數目相同的分割槽中,或是這個數字的幾倍,因為每個分割槽都會執行在不同的執行緒上,如果有太多執行緒,它們之間將變得過於昂貴。

開始:進行基準測試!

我開發了一個 Jupyter 筆記來嘗試使用這個框架,並且釋出在 Github 上,這樣你可以檢視具體資訊甚至是親自執行它。

我執行的基準測試可以在 GitHub 上獲取,這裡列舉了主要內容:

def get_big_mean():
    return dfn.salary.mean().compute()
def get_big_mean_old():
    return df3.salary.mean()

def get_big_max():
    return dfn.salary.max().compute()
def get_big_max_old():
    return df3.salary.max()

def get_big_sum():
    return dfn.salary.sum().compute()
def get_big_sum_old():
    return df3.salary.sum()

def filter_df():
    df = dfn[dfn['salary']>5000]
def filter_df_old():
    df = df3[df3['salary']>5000]
複製程式碼

這是一個有著 2500 萬行的常規 df3,內容是使用來自上一篇文章中的指令碼生成的(從列表中隨機抽取的列名是 name、surname 以及 salary)。我使用了 50 行資料集,並將其連線了 50 萬次,因為我只對它執行所需時間感興趣,對於分析 Per se 卻不感興趣。

dfn 是基於 df3 的 Dask Dataframe。

第一批次的結果:不太樂觀

首先,我嘗試用 3 個分割槽進行測試,因為我只有 4 個核心,所以不想過度使用我的 PC。我用 Dask 的結果不是很理想,而且還必須等待很長時間才能獲取結果,我擔心這可能是因為我做的分割槽太少了:

204.313940048 seconds for get_big_mean
39.7543280125 seconds for get_big_mean_old

131.600986004 seconds for get_big_max
43.7621600628 seconds for get_big_max_old

120.027213097 seconds for get_big_sum
7.49701309204 seconds for get_big_sum_old

0.581165790558 seconds for filter_df
226.700095892 seconds for filter_df_old
複製程式碼

你可以看到,當我是用 Dask 時,大多數操作的速度都要慢得多。這給我了一個提示,那就是我可能不得不使用更多的分割槽。生成延遲評估所花費的數量也是可以忽略不計的(在某些情況下不到半秒),如果我重用它們,就不會隨著時間的推移而攤銷。

我還使用了 apply 方法測試它:

def f(x):
    return (13*x+5)%7

def apply_random_old():
    df3['random']= df3['salary'].apply(f)
    
def apply_random():
    dfn['random']= dfn['salary'].apply(f).compute()
複製程式碼

結果並無差別:

369.541605949 seconds for apply_random
157.643756866 seconds for apply_random_old
複製程式碼

因此,一般情況下,儘管過濾器的速度要快得多,但大多數操作的速度仍然是原來的兩倍。我擔心的是,也許我也應該呼叫 compute 這個函式,所以把這個結果作為對比。

更多分割槽:驚人的速度

再這樣令人沮喪的結果之後,我認為可能是我還沒有使用足夠的分割槽。這樣做的要點是並行執行,或許是我需要更多的並行化?因此我對 8 個分割槽進行了相同的測試,下面是我得到的結果(我忽略了非並行 dataframe,因為它們基本是相同的):

3.08352184296 seconds for get_big_mean
1.3314101696 seconds for get_big_max
1.21639800072 seconds for get_big_sum
0.228978157043 seconds for filter_df

112.135010004 seconds for apply_random
50.2007009983 seconds for value_count_test
複製程式碼

沒錯,大多數操作的執行速度是常規 Dataframe 的 10 倍以上,apply 獲得了更快的速度!我還在 salary 序列上運行了 value_count 方法。對於上下文,請記住,當我在常規的 Dataframe 上執行這個測試時,我等待了 10 分鐘之後,我不得不停止這個過程,這一次只花了 50 秒! 基本上,我只是用錯了工具,而且非常快。比普通的 Dataframes 快得多。

結論

考慮到我在一臺非常舊的 4 核 PC 上,一分鐘內執行 2.5 億行內容,我覺得它會在實際應用中有著舉足輕重的地位。因此我建議,下次你處理本地或從單個 AWS 例項中處理資料集時,可以考慮使用這個框架,它真的非常高效。

我希望你覺得這盤文章有用或者有趣!編寫他所花費的時間超過我的預期,因為一些基準測試花費的時間太長了。記得告訴我在閱讀之前你是否瞭解過 Dask,或者你是否在工作或專案中使用過它。另外,如果有其他更棒的功能,記得告訴我,我並沒有檢測我是否做錯了什麼內容!你的回饋和評論是我寫作的重要原因之一,因為我們都在從中成長。

如果你喜歡這篇文章,可以繼續支援我。可以繼續支援我的寫作。同時你還可以在我這裡瞭解更多 Python 教程、提示和技巧!

如果發現譯文存在錯誤或其他需要改進的地方,歡迎到 掘金翻譯計劃 對譯文進行修改並 PR,也可獲得相應獎勵積分。文章開頭的 本文永久連結 即為本文在 GitHub 上的 MarkDown 連結。


掘金翻譯計劃 是一個翻譯優質網際網路技術文章的社群,文章來源為 掘金 上的英文分享文章。內容覆蓋 AndroidiOS前端後端區塊鏈產品設計人工智慧等領域,想要檢視更多優質譯文請持續關注 掘金翻譯計劃官方微博知乎專欄