加快Python算法的四個方法三數(shù)據(jù)并行化

CDA數(shù)據(jù)分析師（CDA.cn）——真本事，心舒適加快Python算法的四個方法（三）數(shù)據(jù)并行化相信大家在做一些算法經(jīng)常會被龐大的數(shù)據(jù)量所造成的超多計算量需要的時間而折磨的痛苦不已，接下來我們圍繞四個方法來幫助大家加快一下Python的計算時間，減少大家在算法上的等待時間。以下給大家講解關(guān)于數(shù)據(jù)并行化這方面的內(nèi)容。.介紹隨著時間和處理器計算能力的增長，數(shù)據(jù)呈指數(shù)級增長，我們需要找到有效地處理數(shù)據(jù)的方法。那我們應(yīng)該怎么辦呢？GPU是一種非常有效的解決方案。但是，GPU并不是為了機器學(xué)習(xí)而設(shè)計的，它是專門為復(fù)雜的圖像處理和游戲而設(shè)計的。我們使算法能夠在現(xiàn)有GPU上運行，并且確實取得了成果?，F(xiàn)在，谷歌推出了一種名為TPU（張量處理單元）的新設(shè)備，該設(shè)備專門針對TensorFlow上的機器學(xué)習(xí)工作而量身定做的，其結(jié)果確實令人激動。同時英偉達在這方面也并沒有退縮。但是我們將來會在某個時候達到頂峰。即使我們我們現(xiàn)在擁有大量可用的數(shù)據(jù)集，但是單臺機器或計算單元也不足以處理這樣的負載。我們將不得不使用多臺機器來完成我們的任務(wù)。我們將不得不并行化完成我們的任務(wù)。接下來，我們將研究大多數(shù)情況下你將在Python中使用的一些方法。然后再介紹一下Dask和torch.multiprocessing。CDA數(shù)據(jù)分析師(CDA.cn)——真本事，心舒適.^和進程Python庫的Pool和Process方法都來自于multiprocessing它為我們的任務(wù)啟動了一個新的過程，但是方式有所不同。Process每次調(diào)用僅執(zhí)行一個進程：importmultiprocessingasmpp=mp.Process(target=##目標函數(shù)args二##參數(shù)到函數(shù)此調(diào)用將只生產(chǎn)一個進程，該進程將處理在后臺使用給定的參數(shù)處理目標函數(shù)但是這個過程還沒有開始。要啟動它，你必須執(zhí)行以下操作：p.start()現(xiàn)在，你可以將其保留在此處，或者通過以下方式檢查該過程是否完成：p.join()現(xiàn)在它將等待進程完成。不檢查過程是否已完成有許多用途。例如，在客戶端-服務(wù)器應(yīng)用程序中，數(shù)據(jù)包丟失的可能性或無響應(yīng)進程的可能性確實很低，我們可以忽略它，這可以使我們的速度大大提高。[取決于申請程序]對于多個進程，你必須創(chuàng)建多個Process。你想做多少就可以做多少。當你調(diào)用.start()它們時，它們?nèi)慷紝?。processes=[mp.Process(target=func,args=(a,b))for(a,b)inlist]forpinprocesses:p.start()forpinprocesses:p.join()另一方面，Pool啟動固定數(shù)量的進程，然后我們可以為這些進程分配一些任務(wù)。因此，在特定的時間實例中，只有固定數(shù)量的進程將在運行，其余的將在等待狀態(tài)中。進程的數(shù)量通常被選作設(shè)備的內(nèi)核數(shù)，如果此參數(shù)為空，也是可以作為默認的狀態(tài)的。pool=mp.Pool(processes=2)現(xiàn)在有許多方法可以應(yīng)用在Pool。在DataScience中，我們可以避免使用的是Pool.applyfflPool.m叩，因為它們會在任務(wù)完成后立即返回結(jié)果。Pool.apply僅采用一個參數(shù)，并且僅使用一個過程，而Pool.m叩將接受許多參數(shù)，并將其放入我們Pool的過程中。CDA數(shù)據(jù)分析師(CDA.cn)——真本事，心舒適results=[pool.apply(func,(x))forxinX]或者results=pool.map(func,(arg))#僅需要一^個參數(shù)考慮到我們前面的客戶端-服務(wù)器應(yīng)用程序的例子，此處預(yù)定義了要運行的最大進程數(shù)，因此，如果我們有很多請求/數(shù)據(jù)包，則n(僅在Pool中的最大進程)將運行一次，而其他將在等待其中一個進程插槽的隊列中排隊。向量的所有元素的平方我們?nèi)绾问褂脭?shù)據(jù)框A:你可以使用一些可以并行化的函數(shù)df.shape(100,100)dfs=[df.iloc[i*25:i*25+25,0]foriinrange(4)]withPool(4)asp:res=p.map(np.exp,dfs)foriinrange(4):df.iloc[i*25:i*25+25,0]=res[i]它可以方便的對數(shù)據(jù)進行預(yù)處理CDA數(shù)據(jù)分析師（CDA.cn）——真本事，心舒適什么時候使用什么？如果你有很多任務(wù)，但其中很少的任務(wù)是計算密集型的，則應(yīng)使用Process。因為如果它們需要大量計算，它們可能會阻塞你的CPU，并且你的系統(tǒng)可能會崩潰。如果你的系統(tǒng)可以一次處理所有這些操作，那么他們就不必在隊列中等待機會了。并且當你的任務(wù)數(shù)量固定且它們的計算量很大時，應(yīng)使用Pool。因為你同時釋放他們，那么你的系統(tǒng)很可能會崩潰。3.線程處理什么！線程處理在python中進行？python中的線程聲譽。人們的這一點看法是對的。實際上，線程在大多數(shù)情況下是不起作用的。那么問題到底是什么呢？問題就出在GIL（全局解釋器鎖定）上。GIL是在Python的開發(fā)初期就引入的，當時甚至在操作系統(tǒng)中都沒有線程的概念。選擇它是因為它的簡單性。GIL一次僅允許一個CPU進程。也就是說，它一次僅允許一個線程訪問python解釋器。因此，一個線程將整個解釋器Lock,直到它完成。對于單線程程序，它非常快，因為只有一個1。。女要維護。隨著python的流行，有效地推出GIL而不損害所有相關(guān)應(yīng)用程序變得越來越困難。這就是為什么它仍然存在的原因。但是，如果你的任務(wù)不受CPU限制，則仍然可以使用多線程并行（y）。也就是說，如果你的任務(wù)受I/O約束，則可以使用多個線程并獲得加速。因為大多數(shù)時候這些任務(wù)都在等待其他代理（例如磁盤等）的響應(yīng)，并且在這段時間內(nèi)它們可以釋放鎖，而讓其他任務(wù)同時獲取它。4NOTE:（來自于官方網(wǎng)頁）TheGILiscontroversialbecauseitpreventsmultithreadedCPythonprogramsfromtakingfulladvantageofmultiprocessorsystemsincertainsituations.Notethatpotentiallyblockingorlong-runningoperations,suchasI/O,imageprocessing,andNumPynumbercrunching,happenoutsidetheGIL.ThereforeitisonlyinmultithreadedprogramsthatspendalotoftimeinsidetheGIL,interpretingCPythonbytecode,thattheGILbecomesabottleneck.以下是對官方網(wǎng)頁的解釋：GIL是有爭議的，因為它阻止多線程CPython程序在某些情況下充分利用多處理器系統(tǒng)。注意，潛在的阻塞或長時間運行的操作，如I/O、圖像處理和NumPy數(shù)字處理，都發(fā)生在GIL之外。CDA數(shù)據(jù)分析師(CDA.cn)——真本事，心舒適因此，只有在花費大量時間在GIL內(nèi)部解釋CPython字節(jié)碼的多線程程序中，GIL才會成為瓶頸。因此，如果你的任務(wù)受IO限制，例如從服務(wù)器下載一些數(shù)據(jù)，對磁盤進行讀/寫等操作，則可以使用多個線程并獲得加速。fromthreadingimportThreadastimportqueueq=queue.Queue()#用于放置和獲取線程的結(jié)果func_=lambdaq,args:q.put(func(args))threads=[t(target=func_,args=(q,args))forargsinargs_array]fortinthreads:t.start()fortinthreads:t.join()res=[]fortinthreads:res.append(q.get())#這些結(jié)果不一定是按順序排列的要保存線程的結(jié)果，可以使用類似于Queue的方法。為此，你將必須如上所示定義函數(shù)，或者可以在函數(shù)內(nèi)部使用Queue.put(),但是為此，你必須更改函數(shù)定義以QueuW做為參數(shù)?，F(xiàn)在，你在隊列中的結(jié)果不一定是按順序排列的。如果希望結(jié)果按順序排列，則可以傳入一些計數(shù)器作為參數(shù)，如id作為參數(shù)，然后使用這些id來標識結(jié)果的來源。threads=[t(func_,args=(i,q,args))fori,argsinenumerate(args_array)]#并相應(yīng)地更新函數(shù)NOTE:在pandas中的多處理中由于某些原因'read.csv'的方法并沒有提供太多的加速，你可以考慮使用Dask做為替代線程還是進程？一個進程是重量級的，因為它可能包含許多自己的線程(包含至少一個線程)，并且分配了自己的內(nèi)存空間，而線程是輕量級的，因為它在父進程的內(nèi)存區(qū)域上工作，因此制作起來更快。進程內(nèi)的線程之間的通信比較容易，因為它們共享相同的內(nèi)存空間。而進程間的通信(IPC-進程間通信)則比較慢。但是，共享相同數(shù)據(jù)的線程又可能進入競爭狀態(tài)，應(yīng)謹慎使用Locks或使用類似的解決方案。CDA數(shù)據(jù)分析師(CDA.cn)——真本事，心舒適4.DaskDask是一個并行計算庫，它不僅有助于并行化現(xiàn)有的機器學(xué)習(xí)工具(Pandas和Numpy)(即使用高級集合)，而且還有助于并行化低級任務(wù)/功能，并且可以通過制作任務(wù)圖來處理這些功能之間的復(fù)雜交互。[即使用低級調(diào)度程序]這類似于Python的線程或多處理模塊。他們也有一個單獨的機器學(xué)習(xí)庫dask-ml，這與如現(xiàn)有的庫(如sklearn，xgboost和tensorflow)集成在一起。fromdaskimportdelayedasdelay@delaydefadd(x,y):returnx+y@delaydefsq(x):returnx**2現(xiàn)在你可以以任何方式使用這些函數(shù)Dask將使你的執(zhí)行并行化。顧名思義Dask不會立即執(zhí)行函數(shù)調(diào)用，而是根據(jù)對輸入和中間結(jié)果調(diào)用函數(shù)的方式生成計算圖。計算最終結(jié)果pute()Dask在做任何事情的時候都有一種內(nèi)在的并行性。對于如何處理DataFrame的，你可以將其視為分而治之的方法，它將DataFrame分為多個塊，然后并行應(yīng)用給定的函數(shù)。df=dask.DataFrame.read_csv("BigFile.csv",chunks=50000)你的DataFrame已經(jīng)被劃分為了多個塊，你應(yīng)用的每個函數(shù)將分別并行的應(yīng)用所有的模塊。它有大部分的Pandas功能，你可以使用：agg=df.groupby(["column"]).aggregate([“sum","mean"])agg.columns=new_column_namesdf_new=df.merge(agg.reset_index(),on="column",how="left")雖然到目前為止還沒有計算結(jié)果，但是使用omputeC可以并行計算。df_pute().head()它們還具有用于在計算機集群上運行它們的接口。5.torch.multiprocessingtorch.multiprocessing是Pythonmultiprocessing模塊的封裝函數(shù)，其API與原始模塊100%兼容。因此，你可以在此處使用Python的multiprocessing模塊中的Queue',Pipe',Array'CDA數(shù)據(jù)分析師(CDA.cn)——真本事，心舒適等。此外，為了使其更快，他們添加了一個方法，share_memory_()該方法允許數(shù)據(jù)進入一個狀態(tài)，在這個狀態(tài)下任何進程都可以直接使用它，因此將該數(shù)據(jù)作為參數(shù)傳遞給不同的進程不會復(fù)制該數(shù)據(jù)。。你可以共享Tensors，模型的parameters，也可以根據(jù)需要在CPU或GPU上共享它們。來自Pytorch的警告：(關(guān)于GPU上的共享)CUDAAPI要求導(dǎo)出到其他進程的分配在被其他進程使用時仍然有效。你應(yīng)該小心，確保你共享的CUDA張量不會超出范圍，只要有必要。這對于共享模型參數(shù)應(yīng)該不是問題，但是傳遞其他類型的數(shù)據(jù)時應(yīng)該小心。注意，這個限制不適用于共享CPU內(nèi)存。你可以在此處的“PoolandProcess〃部分中使用上面的方法，并且要獲得更快的速度，可以使用share_memory_()方法在所有進程之間共享一個Tensor(例如)而不被需要復(fù)制。#使用多個過程訓(xùn)練一個模型importtorch.multiprocessingasmpdeftrain(model):fordata,labelsindata_loader:optimizer.zero_grad()loss_fn(model(data),labels).backward()optimize