Python数据预处理:使用Dask和Numba并行化加速
摘要:?本文是針對Python設計一種并行處理數據的解決方案——使用Dask和Numba并行化加速運算速度。案例對比分析了幾種不同方法的運算速度,非常直觀,可供參考。
如果你善于使用Pandas變換數據、創建特征以及清洗數據等,那么你就能夠輕松地使用Dask和Numba并行加速你的工作。單純從速度上比較,Dask完勝Python,而Numba打敗Dask,那么Numba+Dask基本上算是無敵的存在。將數值計算分成Numba sub-function和使用Dask map_partition+apply,而不是使用Pandas。對于100萬行數據,使用Pandas方法和混合數值計算創建新特征的速度比使用Numba+Dask方法的速度要慢許多倍。
Python:60.9x | Dask:8.4x | Numba:5.8x |Numba+Dask:1x
作為舊金山大學的一名數據科學碩士,會經常跟數據打交道。使用Apply函數是我用來創建新特征或清理數據的眾多技巧之一。現在,我只是一名數據科學家,而不是計算機科學方面的專家,但我是一個喜歡搗鼓并使得代碼運行更快的程序員。現在,我將會分享我在并行應用上的經驗。
大多Python愛好者可能了解Python實現的全局解釋器鎖(GIL),GIL會占用計算機中所有的CPU性能。更糟糕的是,我們主要的數據處理包,比如Pandas,很少能實現并行處理代碼。
Apply函數vs Multiprocessing.map
Tidyverse已經為處理數據做了一些美好的事情,Plyr是我最喜愛的數據包之一,它允許R語言使用者輕松地并行化他們的數據應用。Hadley Wickham說過:
“plyr是一套處理一組問題的工具:需要把一個大的數據結構分解成一些均勻的數據塊,之后對每一數據塊應用一個函數,最后將所有結果組合在一起。”
對于Python而言,我希望有類似于plyr這樣的數據包可供使用。然而,目前這樣的數據包還不存在,但我可以使用并行數據包構成一個簡單的解決方案。
Dask
之前在Spark上花費了一些時間,因此當我開始使用Dask時,還是比較容易地掌握其重點內容。Dask被設計成能夠在多核CPU上并行處理任務,此外也借鑒了許多Pandas的語法規則。
現在開始本文所舉例子。對于最近的數據挑戰而言,我試圖獲取一個外部數據源(包含許多地理編碼點),并將其與要分析的一大堆街區相匹配。在計算歐幾里得距離的同時,使用最大啟發式將最大值分配給一個街區。
最初的apply:
Dask apply:
dd.from_pandas(my_df,npartitions=nCores).\map_partitions(\lambda df : df.apply(\lambda x : nearest_street(x.lat,x.lon),axis=1)).\compute(get=get) # imports at the end二者看起來很相似,apply核心語句是map_partitions,最后有一個compute()語句。此外,不得不對npartitions初始化。 分區的工作原理就是將Pandas數據幀劃分成塊,對于我的電腦而言,配置是6核-12線程,我只需告訴它使用的是12分區,Dask就會完成剩下的工作。
接下來,將map_partitions的lambda函數應用于每個分區。由于許多數據處理代碼都是獨立地運行,所以不必過多地擔心這些操作的順序問題。最后,compute()函數告訴Dask來處理剩余的事情,并把最終計算結果反饋給我。在這里,compute()調用Dask將apply適用于每個分區,并使其并行處理。
由于我通過迭代行來生成一個新隊列(特征),而Dask apply只在列上起作用,因此我沒有使用Dask apply,以下是Dask程序:
Numba、Numpy和Broadcasting
由于我是根據一些簡單的線性運算(基本上是勾股定理)對數據進行分類,所以認為使用類似下面的Python代碼會運行得更快一些。
Broadcasting用以描述Numpy中對兩個形狀不同的矩陣進行數學計算的處理機制。假設我有一個數組,我會通過迭代并逐個變換每個單元格來改變它
# over one array for cell in array:cell * CONSTANT - CONSTANT2 # over two arrays for i in range(len(array)):array[i] = array[i] + array2[i]
相反,我完全可以跳過for循環,并對整個數組執行操作。Numpy與broadcasting混合使用,用來執行元素智能乘積(對位相乘)。
# over one array (array * CONSTANT) - CONSTANT2 # over two arrays of same length # different lengths follow broadcasting rules array = array - array2Broadcasting可以實現更多的功能,現在看看骨架代碼:
from numba import jit @jit # numba magic def some_func()l3_arr = np.sqrt( (intersections[:,0] - intersections[:,1])**2 +\(intersections[:,2] - intersections[:,3])**2 ) # now l3 is an array containing all of my block lengths # likewise, l1 and l2 are now equal sized arrays # containing distance of point to all intersectionsdist = l1_arr + l2_arrmatch_arr = dist < (l3_arr * 1.2) # so instead of iterating, I just immediately compare all of my # point-to-street distances at once and have a handy # boolean index從本質上講,代碼的功能是改變數組。好的一方面是運行很快,甚至能和Dask并行處理速度比較。其次,如果使用的是最基本的Numpy和Python,那么就可以及時編譯任何函數。壞的一面在于它只適合Numpy和簡單Python語法。我不得不把所有的數值計算從我的函數轉換成子函數,但其計算速度會增加得非常快。
將其一起使用
簡單地使用map_partition()就可以將Numba函數與Dask結合在一起,如果并行操作和broadcasting能夠密切合作以加快運行速度,那么對于大數據集而言,將會看到其運行速度得到大幅提升。
上面的第一張圖表明,沒有broadcasting的線性計算其表現不佳,并行處理和Dask對速度提升也有效果。此外,可以明顯地發現,Dask和Numba組合的性能優于其它方法。
上面的第二張圖稍微有些復雜,其橫坐標是對行數取對數。從第二張圖可以發現,對于1k到10k這樣小的數據集,單獨使用Numba的性能要比聯合使用Numba+Dask的性能更好,盡管在大數據集上Numba+Dask的性能非常好。
優化
為了能夠使用Numba編譯JIT,我重寫了函數以更好地利用broadcasting。之后,重新運行這些函數后發現,平均而言,對于相同的代碼,JIT的執行速度大約快了24%。
可以肯定的說,一定有進一步的優化方法使得執行速度更快,但目前沒有發現。Dask是一個非常友好的工具,本文使用Dask+Numba實現的最好成果是提升運行速度60倍。如果你知道其它的提升執行速度的技巧,歡迎在留言區分享。
文章原標題《Data Pre-Processing in Python: How I learned to love parallelized applies with Dask and Numba》,
作者:Ernest Kim,譯者:海棠,審閱:袁虎。
原文鏈接
干貨好文,請關注掃描以下二維碼:
總結
以上是生活随笔為你收集整理的Python数据预处理:使用Dask和Numba并行化加速的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 从单租户IaaS到多租户PaaS——金融
- 下一篇: 仅使用NumPy完成卷积神经网络CNN的