CDA數(shù)據(jù)分析師 出品

相信大家在做一些算法經(jīng)常會被龐大的數(shù)據(jù)量所造成的超多計算量需要的時間而折磨的痛苦不已，接下來我們圍繞四個方法來幫助大家加快一下Python的計算時間，減少大家在算法上的等待時間。以下給大家講解關(guān)于數(shù)據(jù)并行化這方面的內(nèi)容。

1.介紹

隨著時間和處理器計算能力的增長，數(shù)據(jù)呈指數(shù)級增長，我們需要找到有效地處理數(shù)據(jù)的方法。那我們應(yīng)該怎么辦呢？

GPU是一種非常有效的解決方案。但是，GPU并不是為了機器學習而設(shè)計的，它是專門為復雜的圖像處理和游戲而設(shè)計的。我們使算法能夠在現(xiàn)有GPU上運行，并且確實取得了成果?，F(xiàn)在，谷歌推出了一種名為TPU（張量處理單元）的新設(shè)備，該設(shè)備專門針對TensorFlow上的機器學習工作而量身定做的，其結(jié)果確實令人激動。同時英偉達在這方面也并沒有退縮。

但是我們將來會在某個時候達到頂峰。即使我們我們現(xiàn)在擁有大量可用的數(shù)據(jù)集，但是單臺機器或計算單元也不足以處理這樣的負載。我們將不得不使用多臺機器來完成我們的任務(wù)。我們將不得不并行化完成我們的任務(wù)。

接下來，我們將研究大多數(shù)情況下你將在Python中使用的一些方法。然后再介紹一下Dask和torch.multiprocessing。

2.池和進程

Python庫的Pool和Process方法都來自于multiprocessing它為我們的任務(wù)啟動了一個新的過程，但是方式有所不同。Process每次調(diào)用僅執(zhí)行一個進程：

import multiprocessing as mpp = 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) in list]for p in processes: p.startfor p in processes: 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。在Data Science中，我們可以避免使用的是Pool.apply和Pool.map，因為它們會在任務(wù)完成后立即返回結(jié)果。Pool.apply僅采用一個參數(shù)，并且僅使用一個過程，而Pool.map將接受許多參數(shù)，并將其放入我們Pool的過程中。

results = [pool.apply(func, (x)) for x in X]# 或者 results = pool.map(func, (arg)) # 僅需要一個參數(shù)

考慮到我們前面的客戶端-服務(wù)器應(yīng)用程序的例子，此處預定義了要運行的最大進程數(shù)，因此，如果我們有很多請求/數(shù)據(jù)包，則n（僅在Pool中的最大進程）將運行一次，而其他將在等待其中一個進程插槽的隊列中排隊。

向量的所有元素的平方

# 我們?nèi)绾问褂脭?shù)據(jù)框# A: 你可以使用一些可以并行化的函數(shù)df.shape# (100, 100)dfs = [df.iloc[i*25:i*25+25, 0] for i in range(4)]with Pool(4) as p: res = p.map(np.exp, dfs)for i in range(4): df.iloc[i*25:i*25+25, 0] = res[i]# 它可以方便的對數(shù)據(jù)進行預處理

什么時候使用什么？

如果你有很多任務(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,直到它完成。

對于單線程程序，它非常快，因為只有一個Lock要維護。隨著python的流行，有效地推出GIL而不損害所有相關(guān)應(yīng)用程序變得越來越困難。這就是為什么它仍然存在的原因。

但是，如果你的任務(wù)不受CPU限制，則仍然可以使用多線程并行（y）。也就是說，如果你的任務(wù)受I / O約束，則可以使用多個線程并獲得加速。因為大多數(shù)時候這些任務(wù)都在等待其他代理（例如磁盤等）的響應(yīng)，并且在這段時間內(nèi)它們可以釋放鎖，而讓其他任務(wù)同時獲取它。?

NOTE: (來自于官方網(wǎng)頁)The GIL is controversial because it prevents multithreaded CPython programs from taking full advantage of multiprocessor systems in certain situations. Note that potentially blocking or long-running operations, such as I/O, image processing, and NumPy number crunching, happen outside the GIL. Therefore it is only in multithreaded programs that spend a lot of time inside the GIL, interpreting CPython bytecode, that the GIL becomes a bottleneck.

以下是對官方網(wǎng)頁的解釋：

GIL是有爭議的，因為它阻止多線程CPython程序在某些情況下充分利用多處理器系統(tǒng)。注意，潛在的阻塞或長時間運行的操作，如I/O、圖像處理和NumPy數(shù)字處理，都發(fā)生在GIL之外。因此，只有在花費大量時間在GIL內(nèi)部解釋CPython字節(jié)碼的多線程程序中，GIL才會成為瓶頸。

因此，如果你的任務(wù)受IO限制，例如從服務(wù)器下載一些數(shù)據(jù)，對磁盤進行讀/寫等操作，則可以使用多個線程并獲得加速。

from threading import Thread as timport queueq = queue.Queue # 用于放置和獲取線程的結(jié)果func_ = lambda q, args: q.put(func(args))threads = [t(target=func_, args=(q, args)) for args in args_array]for t in threads: t.startfor t in threads: t.joinres = for t in threads: res.append(q.get) # 這些結(jié)果不一定是按順序排列的

要保存線程的結(jié)果，可以使用類似于Queue 的方法。為此，你將必須如上所示定義函數(shù)，或者可以在函數(shù)內(nèi)部使用Queue.put，但是為此，你必須更改函數(shù)定義以Queue`做為參數(shù)。

現(xiàn)在，你在隊列中的結(jié)果不一定是按順序排列的。如果希望結(jié)果按順序排列，則可以傳入一些計數(shù)器作為參數(shù)，如id作為參數(shù)，然后使用這些id來標識結(jié)果的來源。

threads = [t(func_, args = (i, q, args)) for i, args in enumerate(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或使用類似的解決方案。

4.Dask

Dask是一個并行計算庫，它不僅有助于并行化現(xiàn)有的機器學習工具（Pandas和Numpy）（即使用高級集合），而且還有助于并行化低級任務(wù)/功能，并且可以通過制作任務(wù)圖來處理這些功能之間的復雜交互。[ 即使用低級調(diào)度程序 ]這類似于Python的線程或多處理模塊。

他們也有一個單獨的機器學習庫dask-ml，這與如現(xiàn)有的庫（如sklearn，xgboost和tensorflow）集成在一起。

from dask import delayed as delay@delaydef add(x, y): return x+y@delaydef sq(x): return x**2# 現(xiàn)在你可以以任何方式使用這些函數(shù)，Dask將使你的執(zhí)行并行化。顧名思義，Dask不會立即執(zhí)行函數(shù)調(diào)用，而是根據(jù)對輸入和中間結(jié)果調(diào)用函數(shù)的方式生成計算圖。計算最終結(jié)果:result.compute

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é)果，但是使用.compute可以并行計算。df_new.compute.head

它們還具有用于在計算機集群上運行它們的接口。

5.torch.multiprocessing

torch.multiprocessing是Python multiprocessing模塊的封裝函數(shù)，其API與原始模塊100％兼容。因此，你可以在此處使用Python的 multiprocessing模塊中的Queue'，Pipe'，Array'等。此外，為了使其更快，他們添加了一個方法，share_memory_該方法允許數(shù)據(jù)進入一個狀態(tài)，在這個狀態(tài)下任何進程都可以直接使用它，因此將該數(shù)據(jù)作為參數(shù)傳遞給不同的進程不會復制該數(shù)據(jù)。 。

你可以共享Tensors，模型的parameters，也可以根據(jù)需要在CPU或GPU上共享它們。

來自Pytorch的警告：（關(guān)于GPU上的共享） CUDA API要求導出到其他進程的分配在被其他進程使用時仍然有效。你應(yīng)該小心，確保你共享的CUDA張量不會超出范圍，只要有必要。這對于共享模型參數(shù)應(yīng)該不是問題，但是傳遞其他類型的數(shù)據(jù)時應(yīng)該小心。注意，這個限制不適用于共享CPU內(nèi)存。

你可以在此處的"Pool and Process"部分中使用上面的方法，并且要獲得更快的速度，可以使用share_memory_方法在所有進程之間共享一個Tensor（例如）而不被需要復制。

# 使用多個過程訓練一個模型:import torch.multiprocessing as mpdef train(model): for data, labels in data_loader: optimizer.zero_grad loss_fn(model(data), labels).backward optimizer.step # 這將更新共享參數(shù)model = nn.Sequential(nn.Linear(n_in, n_h1), nn.ReLU, nn.Linear(n_h1, n_out))model.share_memory #需要"fork"方法工作processes = for i in range(4): # NO.的過程 p = mp.Process(target=train, args=(model,)) p.start processes.append(p)for p in processes: p.join

下一期繼續(xù)看加快Python算法的第4種方法——Dask！