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

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