CDA數據分析師 出品

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

1.介紹

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

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

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

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

2.池和進程

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

import multiprocessing as mpp = mp.Process(target= ##目標函數, args= ##參數到函數)# 此調用將只生產一個進程，該進程將處理在后臺使用給定的參數處理目標函數

但是這個過程還沒有開始。要啟動它，你必須執行以下操作：

p.start

現在，你可以將其保留在此處，或者通過以下方式檢查該過程是否完成：

p.join#現在它將等待進程完成。

不檢查過程是否已完成有許多用途。例如，在客戶端-服務器應用程序中，數據包丟失的可能性或無響應進程的可能性確實很低，我們可以忽略它，這可以使我們的速度大大提高。[取決于申請程序]

對于多個進程，你必須創建多個Process。你想做多少就可以做多少。當你調用.start它們時，它們全部都將會啟動。

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啟動固定數量的進程，然后我們可以為這些進程分配一些任務。因此，在特定的時間實例中，只有固定數量的進程將在運行，其余的將在等待狀態中。進程的數量通常被選作設備的內核數，如果此參數為空，也是可以作為默認的狀態的。

pool = mp.Pool(processes=2)

現在有許多方法可以應用在Pool。在Data Science中，我們可以避免使用的是Pool.apply和Pool.map，因為它們會在任務完成后立即返回結果。Pool.apply僅采用一個參數，并且僅使用一個過程，而Pool.map將接受許多參數，并將其放入我們Pool的過程中。

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

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

向量的所有元素的平方

# 我們如何使用數據框# A: 你可以使用一些可以并行化的函數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]# 它可以方便的對數據進行預處理

什么時候使用什么？

如果你有很多任務，但其中很少的任務是計算密集型的，則應使用Process。因為如果它們需要大量計算，它們可能會阻塞你的CPU，并且你的系統可能會崩潰。如果你的系統可以一次處理所有這些操作，那么他們就不必在隊列中等待機會了。

并且當你的任務數量固定且它們的計算量很大時，應使用Pool。因為你同時釋放他們，那么你的系統很可能會崩潰。

3.線程處理

什么！線程處理在python中進行？

python中的線程聲譽。人們的這一點看法是對的。實際上，線程在大多數情況下是不起作用的。那么問題到底是什么呢？

問題就出在GIL（全局解釋器鎖定）上。GIL是在Python的開發初期就引入的，當時甚至在操作系統中都沒有線程的概念。選擇它是因為它的簡單性。

GIL一次僅允許一個CPU進程。也就是說，它一次僅允許一個線程訪問python解釋器。因此，一個線程將整個解釋器Lock,直到它完成。

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

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

NOTE: (來自于官方網頁)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.

以下是對官方網頁的解釋：

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

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

from threading import Thread as timport queueq = queue.Queue # 用于放置和獲取線程的結果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) # 這些結果不一定是按順序排列的

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

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

threads = [t(func_, args = (i, q, args)) for i, args in enumerate(args_array)]# 并相應地更新函數NOTE:在pandas中的多處理中由于某些原因 'read.csv' 的方法并沒有提供太多的加速，你可以考慮使用Dask做為替代

線程還是進程？

一個進程是重量級的，因為它可能包含許多自己的線程（包含至少一個線程），并且分配了自己的內存空間，而線程是輕量級的，因為它在父進程的內存區域上工作，因此制作起來更快。

進程內的線程之間的通信比較容易，因為它們共享相同的內存空間。而進程間的通信（IPC-進程間通信）則比較慢。但是，共享相同數據的線程又可能進入競爭狀態，應謹慎使用Locks或使用類似的解決方案。

4.Dask

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

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

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

Dask在做任何事情的時候都有一種內在的并行性。對于如何處理DataFrame的，你可以將其視為分而治之的方法，它將DataFrame分為多個塊，然后并行應用給定的函數。

df = dask.DataFrame.read_csv("BigFile.csv", chunks=50000)# 你的DataFrame已經被劃分為了多個塊，你應用的每個函數將分別并行的應用所有的模塊。它有大部分的Pandas功能，你可以使用：agg = df.groupby(["column"]).aggregate(["sum", "mean"])agg.columns = new_column_namesdf_new = df.merge(agg.reset_index, on="column", how="left")# 雖然到目前為止還沒有計算結果，但是使用.compute可以并行計算。df_new.compute.head

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

5.torch.multiprocessing

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

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

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

你可以在此處的"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 # 這將更新共享參數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

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