Uber 開發(fā)了 POET、Go-Explore 和 GTN 等算法，這些算法利用大量的計算來訓(xùn)練神經(jīng)網(wǎng)絡(luò)模型。為了使未來幾代類似算法的大規(guī)模計算成為可能，Uber 進而開發(fā)了一種新的分布式計算庫 Fiber，它可以幫助用戶輕松地將本地計算方法擴展到成百上千臺機器上。Fiber 可以使使用 Python 的大規(guī)模計算項目變得快速、簡單和資源高效，從而簡化 ML 模型訓(xùn)練過程，并獲得更優(yōu)的結(jié)果。

　　本文最初發(fā)布于 Uber 工程博客，由 InfoQ 中文站翻譯并分享。

　　項目地址：https://github.com/uber/fiber

　　在過去的幾年中，計算機不斷增強的處理能力推動了機器學(xué)習(xí)的進步。算法越來越多地利用并行性，并依賴分布式訓(xùn)練來處理大量數(shù)據(jù)。然而，隨之而來的是增加數(shù)據(jù)和訓(xùn)練的需求，這對管理和利用大規(guī)模計算資源的軟件提出了巨大的挑戰(zhàn)。

　　在 Uber，我們開發(fā)了POET、Go-Explore和GTN等算法，這些算法利用大量的計算來訓(xùn)練神經(jīng)網(wǎng)絡(luò)模型。為了使未來幾代類似算法的大規(guī)模計算成為可能，我們開發(fā)了一種新的分布式計算庫Fiber，它可以幫助用戶輕松地將本地計算方法擴展到成百上千臺機器上。Fiber 可以使使用 Python 的大規(guī)模計算項目變得快速、簡單和資源高效，從而簡化 ML 模型訓(xùn)練過程，并獲得更優(yōu)的結(jié)果。

　　大規(guī)模分布式計算的挑戰(zhàn)

　　在理想情況下，將運行在一臺機器上的應(yīng)用程序擴展為運行在一批機器上的應(yīng)用程序應(yīng)該很容易，只需更改命令行參數(shù)即可。然而，在現(xiàn)實世界中，這并不容易。

　　我們每天都與許多運行大規(guī)模分布式計算任務(wù)的人一起工作，我們發(fā)現(xiàn)，現(xiàn)在很難利用分布式計算的原因有以下幾個：

• 在筆記本或臺式機本地運行代碼與在生產(chǎn)集群上運行代碼之間存在著巨大的差距。你可以讓MPI在本地運行，但在計算機集群上運行它是完全不同的過程。

• 不能動態(tài)擴展。如果你啟動了一個需要大量資源的作業(yè)，那么你很可能需要等待，直到所有資源都分配好了才可以運行該作業(yè)。這個等待降低了擴展的效率。

• 錯誤處理缺失。在運行時，有些作業(yè)可能會失敗。你可能不得不還原部分結(jié)果或整個地放棄本次運行。

• 學(xué)習(xí)成本很高。每個系統(tǒng)都有不同的 API 和編程約定。要使用新系統(tǒng)啟動作業(yè)，用戶必須學(xué)習(xí)一套全新的約定。

　　新的 Fiber 平臺專門解決了這些問題。它為更廣泛的用戶群體提供了無縫使用大規(guī)模分布式計算的可能。

　　Fiber 簡介

　　Fiber 是一個用于現(xiàn)代計算機集群的基于 Python 的分布式計算庫。用戶可以利用這個系統(tǒng)針對整個計算機集群進行編程，而不是只針對單個臺式機或筆記本電腦。它最初是為了支持像POET這樣的大規(guī)模并行科學(xué)計算項目而開發(fā)的，Uber 也已經(jīng)用它來支持類似的項目。Fiber 的功能非常強大，這樣主要是因為：

• 易于使用。Fiber 允許用戶編寫在計算機集群上運行的程序，而不需要深入研究計算機集群的細節(jié)。

• 易于學(xué)習(xí)。Fiber 提供了與 Python 標(biāo)準(zhǔn)多處理庫相同的 API。知道如何使用多處理庫的工程師可以很容易地用 Fiber 編寫計算機集群程序。

• 快速可靠。Fiber 的通信中樞基于Nanomsg構(gòu)建，這是一個高性能異步消息傳遞庫，可以提供快速、可靠的通信。

• 不需要部署。Fiber 在計算機集群上的運行方式與普通應(yīng)用程序相同。它會自動為用戶處理資源分配和通信。

• 提供了可靠的計算。Fiber 內(nèi)置的錯誤處理功能讓用戶可以專注于編寫實際的應(yīng)用程序代碼，而不是處理崩潰問題。當(dāng)運行一個工作進程池時，這尤其有價值。

　　除了這些好處之外，F(xiàn)iber 還可以在特別關(guān)注性能的領(lǐng)域與其他專用框架搭配使用。例如，對于隨機梯度下降（SGD），F(xiàn)iber 的Ring 特性可以幫助我們在計算機集群上建立分布式訓(xùn)練作業(yè)，并允許它與Horovod或torch.distributed協(xié)同。

圖 1：Fiber 啟動許多不同的作業(yè)支持（job-backed）進程，然后在其中運行不同的 Fiber 組件和用戶進程。Fiber Master 是管理所有其他進程的主進程。有些進程（如 Ring Node）保持成員之間的通信。

　　Fiber 可以幫助從事大規(guī)模分布式計算的用戶減少從產(chǎn)生想法到在計算集群上實際運行分布式作業(yè)的時間。它還可以幫助用戶屏蔽配置和資源分配任務(wù)的繁瑣細節(jié)，并且可以縮短調(diào)試周期，簡化從本地開發(fā)到集群開發(fā)的轉(zhuǎn)換。

　　架構(gòu)

　　Fiber 讓我們可以靈活地為經(jīng)典的多處理 API 選擇可以在不同集群管理系統(tǒng)上運行的后端。為了實現(xiàn)這種集成，F(xiàn)iber 被分為三個不同的層：API 層、后端層和集群層。API 層為 Fiber 提供了進程、隊列、池和管理器等基本構(gòu)建塊。它們具有與多處理相同的語義，但是我們對它們進行擴展了，使它們可以在分布式環(huán)境中工作。后端層處理在不同集群管理器上創(chuàng)建或終止作業(yè)的任務(wù)。當(dāng)用戶新增一個后端時，所有其他 Fiber 組件（隊列、池等）都不需要更改。最后，集群層由不同的集群管理器組成。盡管它們不是 Fiber 本身的一部分，但是它們幫助 Fiber 管理資源并跟蹤不同的作業(yè)，減少了 Fiber 所需要跟蹤的項的數(shù)量。圖 2 是總體架構(gòu)圖：

圖 2：Fiber 的架構(gòu)包括 API 層、后端層和集群層，這讓它可以在不同的集群管理系統(tǒng)上運行。

　　作業(yè)支持進程

　　Fiber 引入了一個新的概念，稱為作業(yè)支持過程（也稱為 Fiber 進程）。這些進程與 Python 多處理庫中的進程類似，但是更靈活：多處理庫中的進程只在本地機器上運行，但 Fiber 進程可以在不同的機器上遠程運行，也可以在同一機器上本地運行。當(dāng)新的 Fiber 進程啟動時，F(xiàn)iber 會在當(dāng)前計算機集群上創(chuàng)建一個具有適當(dāng) Fiber 后端的新作業(yè)。

圖 3：Fiber 中的每個作業(yè)支持進程都是在計算機集群上運行的一個容器化作業(yè)。每個作業(yè)支持進程也有自己的 CPU、GPU 和其他計算資源。在容器內(nèi)運行的代碼是自包含的。

　　Fiber 使用容器來封裝當(dāng)前進程的運行環(huán)境（如上圖 3 所示），其中包括所有必需的文件、輸入數(shù)據(jù)和其他依賴的程序包，而且要保證每個元素都是自包含的。所有子進程都以與父進程相同的容器鏡像啟動，以確保運行環(huán)境的一致性。因為每個進程都是一個集群作業(yè)，所以它的生命周期與集群上的任何作業(yè)相同。為了方便用戶，F(xiàn)iber 被設(shè)計成直接與計算機集群管理器交互。因此，不像Apache Spark或ipyparallel，F(xiàn)iber 不需要在多臺機器上設(shè)置，也不需要通過任何其他機制引導(dǎo)。它只需要作為一個普通的 Python pip 包安裝在一臺機器上。

　　組件

　　Fiber 基于 Fiber 進程實現(xiàn)了大多數(shù)多處理 API，包括管道、隊列、池和管理器。

　　Fiber 中隊列和管道的行為方式與多處理相同。不同之處在于，F(xiàn)iber 中的隊列和管道由運行在不同機器上的多個進程共享。兩個進程可以從同一個管道讀取和寫入數(shù)據(jù)。此外，隊列可以在不同機器上的多個進程之間共享，每個進程可以同時向同一隊列發(fā)送或從同一隊列接收信息。Fiber 隊列是用高性能異步消息隊列系統(tǒng) Nanomsg 實現(xiàn)的。

圖 4：Fiber 可以在不同的 Fiber 進程之間共享隊列。在本例中，一個 Fiber 進程與隊列位于同一臺機器上，另外兩個進程位于另一臺機器上。一個進程寫入隊列，另外兩個進程讀取隊列。

　　Fiber 也支持池，如下圖 5 所示。它們讓用戶可以管理工作進程池。Fiber 使用作業(yè)支持進程擴展池，以便每個池可以管理數(shù)千個（遠程）工作進程。用戶還可以同時創(chuàng)建多個池。

圖 5：在具有三個工作進程的池中，如本例所示，兩個工作進程位于一臺機器上，另一個位于另一臺機器上。它們共同處理提交到主進程中任務(wù)隊列的任務(wù)，并將結(jié)果發(fā)送到結(jié)果隊列。

　　管理器和代理對象使 Fiber 能夠支持共享存儲，這在分布式系統(tǒng)中至關(guān)重要。通常，這個功能由計算機集群外部存儲系統(tǒng)如 Cassandra 和 Redis 提供。相反，F(xiàn)iber 為應(yīng)用程序提供了內(nèi)置的內(nèi)存存儲。該接口與多處理系統(tǒng)中的管理器類型接口相同。

　　Ring 是對多處理 API 的擴展，可以用于分布式計算設(shè)置。在 Fiber 中，Ring 指的是一組共同工作的、相對平等的進程。與池不同，Ring 沒有主進程和輔助進程的概念。Ring 內(nèi)的所有成員承擔(dān)大致相同的責(zé)任。Fiber 的 Ring 模型拓撲（如下圖 6 所示）在機器學(xué)習(xí)分布式 SGD 中非常常見，torch.distributed和Horovod就是例子。一般來說，在一個計算機集群上啟動這種工作負載是非常具有挑戰(zhàn)性的；Fiber 提供 Ring 特性就是為了幫助建立這樣的拓撲。

圖 6：在一個有四個節(jié)點的 Fiber Ring 中，Ring 節(jié)點 0 和 Ring 節(jié)點 3 運行在同一臺機器上，但在兩個不同的容器中。Ring 節(jié)點 1 和節(jié)點 2 都在單獨的機器上運行。所有這些進程共同運行同一函數(shù)的副本，并在運行期間相互通信。

　　應(yīng)用程序

　　借助上述靈活的組件，我們現(xiàn)在可以使用 Fiber 構(gòu)建應(yīng)用程序了。在這一節(jié)中，我們將展示兩種使用 Fiber 幫助用戶構(gòu)建分布式應(yīng)用程序的方式。

　　賦能新應(yīng)用程序

　　在下面的例子中，我們將展示工程師如何運用 Fiber 來實現(xiàn)大規(guī)模分布式計算。這個例子演示的是一個強化學(xué)習(xí)（RL）算法。通常，分布式 RL 的通信模式涉及在機器之間發(fā)送不同類型的數(shù)據(jù)，包括動作、神經(jīng)網(wǎng)絡(luò)參數(shù)、梯度、per-step/episode 觀察及獎勵。

　　Fiber 實現(xiàn)了管道和池來傳輸這些數(shù)據(jù)。在底層，池是普通的 Unix 套接字，為使用 Fiber 的應(yīng)用程序提供接近線路速度的通信。現(xiàn)代計算機網(wǎng)絡(luò)的帶寬通常高達每秒幾百千兆。通過網(wǎng)絡(luò)傳輸少量數(shù)據(jù)通常速度很快。

　　此外，如果有許多不同的進程向一個進程發(fā)送數(shù)據(jù)，進程間通信延遲也不會增加太多，因為數(shù)據(jù)傳輸可以并行進行。事實證明，F(xiàn)iber 池可以作為許多 RL 算法的基礎(chǔ)，因為模擬器可以在各個池工作進程中運行，并且結(jié)果可以并行回傳。

　　下面的示例顯示了使用 Fiber 實現(xiàn)的簡化 RL 代碼：

# fiber.BaseManager is a manager that runs remotely
class RemoteEnvManager (fiber.managers.AsyncManager):
pass
class Env (gym.env):
# gym env
pass
RemoteEnvManager.register (‘Env’, Env)
def build_model ():
# create a new policy model
return model
def update_model (model, observations):
# update model with observed data
return new_model
def train ():
model = build_model ()
manager = RemoteEnvManager ()
num_envs = 10
envs = [manager.Env () for i in range (num_envs)]
handles = [envs[i].reset () for i in num_envs]
obs = [handle.get () for handle in handles]
for i in range (1000):
actions = model (obs)
handles = [env.step () for action in actions]
obs = [handle.get () for handle in handles]
model = update_model (model, obs)

　　賦能現(xiàn)有的多處理應(yīng)用程序

　　許多 Python 用戶利用了多處理。Fiber 為此類應(yīng)用程序提供了更多的機會，通過這種系統(tǒng)，只需更改幾行代碼，就可以在類似于 Kubernetes 的計算機集群上的分布式設(shè)置中運行。

　　例如，OpenAI Baselines是一個非常流行的 RL 庫，它有許多參考算法，比如DQN和PPO。它的缺點是只能在一臺機器上工作。如果希望大規(guī)模地訓(xùn)練 PPO，就必須創(chuàng)建自己的基于 MPI 的系統(tǒng)并手動設(shè)置集群。

　　相比之下，有了 Fiber，事情就簡單多了。它可以無縫地擴展像 PPO 這樣的 RL 算法，從而利用分布式環(huán)境的數(shù)百個工作進程。Fiber 提供了與多處理相同的 API，OpenAI Baselines 就是使用這些 API 在本地獲取多核 CPU 的處理能力。要讓 OpenAI Baselines 使用 Fiber，只需要修改一行代碼：

　　修改完這行代碼，OpenAI Baselines 就可以在 Kubernetes 上運行了。我們在這里提供了在 Kubernetes 上運行 OpenAI Baselines 的完整指南。

　　錯誤處理

　　Fiber 實現(xiàn)了基于池的錯誤處理。在創(chuàng)建新池時，還將創(chuàng)建關(guān)聯(lián)的任務(wù)隊列、結(jié)果隊列和掛起表。然后，用戶可以將新創(chuàng)建的任務(wù)添加到任務(wù)隊列中。該任務(wù)隊列由主進程和工作進程共享。每個工作進程從任務(wù)隊列中獲取一個任務(wù)，然后在該任務(wù)中運行任務(wù)函數(shù)。每當(dāng)用戶從任務(wù)隊列中刪除一個任務(wù)時，F(xiàn)iber 就會在掛起表中添加一個條目。工作進程完成該任務(wù)后會將結(jié)果放入結(jié)果隊列中。然后，F(xiàn)iber 從掛起表中刪除與該任務(wù)相關(guān)的條目。

圖 7：上圖是一個包含四個工作進程的普通 Fiber 池。在下圖，Worker 3 出現(xiàn)故障，因此 Fiber 啟動一個新的工作進程（Worker 5），然后準(zhǔn)備將其添加到池中。

　　如果池里有一個工作進程在處理過程中失敗，如上圖 7 所示，父池作為所有工作進程的進程管理器將會檢測到該失敗。然后，如果這個失敗的進程有掛起任務(wù)，則父池會將掛起表中的掛起任務(wù)放回到任務(wù)隊列中。接下來，它啟動一個新的工作進程來替換之前失敗的進程，并將新創(chuàng)建的工作進程綁定到任務(wù)隊列和結(jié)果隊列。

　　性能

　　Fiber 最重要的應(yīng)用之一是擴展計算算法（如 RL) 和基于群體的方法（如 ES）。在這些應(yīng)用程序中，延遲非常關(guān)鍵。RL 和基于群體的方法經(jīng)常應(yīng)用于需要與模擬器（如ALE、Gym和Mujoco）頻繁交互以評估策略和收集經(jīng)驗的設(shè)置中。等待模擬器結(jié)果所帶來的延遲嚴重影響了整體的訓(xùn)練性能。

　　為了測試 Fiber，我們將其性能與其他框架進行了比較。我們還在框架開銷測試中增加了Ray，以提供一些初步結(jié)果，并希望在將來添加更詳細的結(jié)果。

　　通常有兩種方法可以減少 RL 算法和基于群體的方法的延遲。要么我們可以減少需要傳輸?shù)臄?shù)據(jù)量，要么我們可以提升不同進程之間通信通道的速度。為了加快通信處理，F(xiàn)iber 使用 Nanomsg 實現(xiàn)了管道和池。此外，用戶還可以使用speedus這樣的庫進一步提高性能。

　　框架開銷

　　通常，框架中的組件會影響計算資源，因此，我們測試了 Fiber 的開銷。我們比較了 Fiber、Python 多處理庫、Apache Spark、Ray 和 ipyparallel。在測試過程中，我們創(chuàng)建了一批工作負載，完成這些任務(wù)所需的總時間是固定的。每個任務(wù)的持續(xù)時間從 1 秒到 1 毫秒不等。

　　對于每個框架，我們在本地運行了 5 個工作進程，并通過調(diào)整批次的大小來確保每個框架的總耗時大約為 1 秒（即 1 毫秒的任務(wù)，我們運行了 5000 個）。我們假設(shè)，F(xiàn)iber 的性能應(yīng)該和多處理差不多，因為 Fiber 和多處理都不依賴于復(fù)雜的調(diào)度機制。相反，我們認為 Apache Spark、Ray 和 ipyparallel 會比 Fiber 慢，因為它們中間依賴于調(diào)度器。

圖 8：在測試 Fiber、Python 多處理庫、Apache Spark 和 ipyprallel 的框架開銷時，我們在本地運行了 5 個工作進程，并調(diào)整批次大小，使每個框架在大約 1 秒鐘內(nèi)完成任務(wù)。

　　當(dāng)任務(wù)持續(xù)時間為 100 毫秒或更多時，F(xiàn)iber 幾乎沒有表現(xiàn)出任何差異，當(dāng)任務(wù)持續(xù)時間降至 10 毫秒或 1 毫秒時，它比其他框架更接近多處理庫。

　　我們以多處理作為參考，因為它非常輕量級，除了創(chuàng)建新進程和并行運行任務(wù)外沒有實現(xiàn)任何其他特性。此外，它還利用了僅在本地可用的通信機制（例如共享內(nèi)存、Unix 域套接字等）。這使得支持分布式資源管理系統(tǒng)的其他框架難以超越多處理，因為這些系統(tǒng)無法利用類似的機制。

圖 9：我們的開銷測試顯示，F(xiàn)iber 的執(zhí)行情況與 Python 多處理庫類似，在 1 毫秒處，ipyparallel 和 Apache Spark 處理任務(wù)的耗時更長。最佳完成時間為 1 秒。

　　與 Fiber 相比，ipyparallel 和 Apache Spark 在每個任務(wù)持續(xù)時間上都落后很多。當(dāng)任務(wù)持續(xù)時間為 1 毫秒時，ipyparallel 花費的時間幾乎是 Fiber 的 24 倍，Apache Spark 花費的時間是后者的 38 倍。顯然，當(dāng)任務(wù)持續(xù)時間較短時，ipyparallel 和 Apache Spark 都引入了相當(dāng)大的開銷，而且，對于 RL 和基于群體的方法，它們不如 Fiber 合適，后者使用了模擬器，響應(yīng)時間只有幾毫秒。我們還可以看到，在運行 1 毫秒的任務(wù)時，Ray 花費的時間大約是 Fiber 的 2.5 倍。

　　分布式任務(wù)測試

　　為了探究 Fiber 的可伸縮性和效率，我們將其與 ipyparallel 進行了比較，由于之前的性能測試結(jié)果，我們沒有考慮 Apache Spark。我們也排除了 Python 多處理庫，因為它不能擴展到一臺機器之外。我們運行了 50 次進化策略迭代（ES），根據(jù)它們的耗時對比了 Fiber 和 ipyparallel 的可伸縮性和效率。

　　在工作負載相同的情況下，我們預(yù)計 Fiber 可以完成得更快，因為前面已測試過，它的開銷比 ipyparallel 小得多。對于 Fiber 和 ipyparallel，我們使用的群體大小為 2048，因此，無論工作進程的數(shù)量多少，總計算量都是固定的。我們還在兩者中實現(xiàn)了相同的共享噪音表，每八個工作進程共享一個噪音表。這項工作的實驗域是OpenAI GymBipedal Walker Hardcore 環(huán)境的一個修改版本，這里對修改進行了描述。

圖 10：當(dāng) ES 迭代 50 次以上時，使用不同數(shù)量的工作進程運行 ES，F(xiàn)iber 的擴展性均優(yōu)于 ipyparallel。每個工作進程在單個 CPU 上運行。

　　主要結(jié)果是，F(xiàn)iber 的擴展性比 ipyparallel 更好，并且完成每次測試的速度明顯更快。隨著工作進程數(shù)從 32 增加到 1024，F(xiàn)iber 的運行時間逐漸縮短。相比之下，當(dāng)工作進程數(shù)從從 256 增加到 512 時，ipyparallel 的運行時間逐漸變長。在使用 1024 個工作進程時，由于進程之間的通信錯誤，ipyparallel 未能完成運行。這個失敗削弱了 ipyparallel 運行大規(guī)模并行計算的能力。根據(jù)Amdahl 定律，我們看到，當(dāng)工作進程數(shù)增加到 512 以上時，F(xiàn)iber 的收益會減少。在這種情況下，主進程處理數(shù)據(jù)的速度就會成為瓶頸。

　　總的來說，在所有工作進程數(shù)的測試中，F(xiàn)iber 的性能都超過了 ipyparallel。此外，與 ipyparallel 不同的是，F(xiàn)iber 在運行 1024 個工作進程時也完成了這項工作。這個結(jié)果更能顯示出 Fiber 與 ipyparallel 相比具有更好的可伸縮性。

　　結(jié)論

　　Fiber 是一個新的 Python 分布式庫，現(xiàn)已開源。我們設(shè)計它是為了讓用戶能夠在一個計算機集群上輕松地實現(xiàn)大規(guī)模計算。實驗表明，F(xiàn)iber 實現(xiàn)了我們的許多目標(biāo)，包括有效地利用大量的異構(gòu)計算硬件，動態(tài)地伸縮算法以提高資源使用效率，以及減少在計算機集群上運行復(fù)雜算法所需的工程負擔(dān)。

　　我們希望，F(xiàn)iber 將進一步加快解決工程難題的進展，使開發(fā)方法并大規(guī)模地運行以了解其好處變得更容易。要了解更多細節(jié)，請查看Fiber GitHub 庫。

　　查看英文原文：

　　https://eng.uber.com/fiberdistributed/

總結(jié)

以上是生活随笔為你收集整理的Uber正式开源分布式机器学习平台Fiber的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。