在 LinkedIn，我們非常依賴離線數(shù)據(jù)分析來進(jìn)行數(shù)據(jù)驅(qū)動的決策。多年來，Apache Spark 已經(jīng)成為 LinkedIn 的主要計(jì)算引擎，以滿足這些數(shù)據(jù)需求。憑借其獨(dú)特的功能，Spark 為 LinkedIn 的許多關(guān)鍵業(yè)務(wù)提供支持，包括數(shù)據(jù)倉庫、數(shù)據(jù)科學(xué)、AI/ML、A/B 測試和指標(biāo)報(bào)告。需要大規(guī)模數(shù)據(jù)分析的用例數(shù)量也在快速增長。從 2017 年到現(xiàn)在，LinkedIn 的 Spark 使用量同比增長了大約 3 倍。因此，LinkedIn 的 Spark 引擎現(xiàn)在運(yùn)行在一個(gè)龐大的基礎(chǔ)設(shè)施之上。我們的生產(chǎn)集群有 10，000 多個(gè)節(jié)點(diǎn)，Spark 作業(yè)現(xiàn)在占集群計(jì)算資源使用量的 70% 以上，每日處理的數(shù)據(jù)量達(dá)數(shù)十 PB。解決擴(kuò)展性挑戰(zhàn)以確保我們的 Spark 計(jì)算基礎(chǔ)設(shè)施可持續(xù)發(fā)展是 LinkedIn Spark 團(tuán)隊(duì)工作的核心。

盡管 Apache Spark 有很多好處，這使得它在 LinkedIn 和整個(gè)行業(yè)中都很受歡迎，但在我們的數(shù)據(jù)規(guī)模下使用 Spark 時(shí)，我們?nèi)匀挥龅搅艘恍┨魬?zhàn)。正如我們在 Spark + AI Summit 2020 演講中所述，這些挑戰(zhàn)涉及多個(gè)層面，包括計(jì)算資源管理、計(jì)算引擎可擴(kuò)展性和用戶生產(chǎn)率。我們將在這篇博文中關(guān)注 Spark shuffle 可擴(kuò)展性挑戰(zhàn)，并介紹 Magnet，一種新穎的基于推送模式的 shuffle 服務(wù)（push-based shuffle service）。

Shuffle 基礎(chǔ)知識

數(shù)據(jù) shuffle 是 MapReduce 計(jì)算范式中的一項(xiàng)重要操作，為 Apache Spark 和許多其他現(xiàn)代大數(shù)據(jù)計(jì)算引擎提供動力。shuffle 操作基本上是通過相應(yīng)階段的 map 和 reduce 任務(wù)之間的 all-to-all 連接來傳輸中間數(shù)據(jù)。通過 shuffle，數(shù)據(jù)將根據(jù)每個(gè)記錄的分區(qū)鍵中的值在所有的 shuffle 分區(qū)上進(jìn)行適當(dāng)?shù)姆謪^(qū)。如圖 1 所示：

雖然 shuffle 操作的基本概念很簡單，但不同的計(jì)算引擎采用了不同的方法來實(shí)現(xiàn)它。在 LinkedIn，我們在 Apache YARN 之上運(yùn)行 Spark，并利用 Spark 的External Shuffle Service （ESS） 來進(jìn)行 shuffle 操作。如圖 2 所示：

通過這種設(shè)置，計(jì)算集群中的每個(gè)節(jié)點(diǎn)都部署了一個(gè) Spark ESS 實(shí)例。當(dāng) Spark executors 運(yùn)行 shuffle 的 mapper 任務(wù)時(shí)，這些任務(wù)會在本地磁盤上生成 shuffle 文件。每個(gè) shuffle 文件由多個(gè) shuffle 塊組成，每個(gè)塊代表屬于相應(yīng) shuffle 分區(qū)的數(shù)據(jù)。生成這些 shuffle 文件后，本地 Spark ESS 實(shí)例知道在哪里可以找到這些 shuffle 文件以及不同 mapper 任務(wù)生成的各個(gè) shuffle 塊。當(dāng) Spark 執(zhí)行器開始運(yùn)行同一個(gè) shuffle 的 reducer 任務(wù)時(shí)，這些任務(wù)將從 Spark driver 獲取有關(guān)在哪里可以找到 shuffle 塊的信息作為它們的任務(wù)輸入。然后這些 reducer 會向遠(yuǎn)程 Spark ESS 實(shí)例發(fā)送請求，嘗試獲取它們對應(yīng)的 shuffle 塊。Spark ESS 收到此類請求后，將從本地磁盤讀取相應(yīng)的 shuffle 塊并將數(shù)據(jù)發(fā)送回 reducer。

挑戰(zhàn)

Spark 現(xiàn)有的 shuffle 機(jī)制在性能和容錯(cuò)要求之間取得了很好的平衡。然而，當(dāng)在我們的規(guī)模下運(yùn)行 Spark shuffle 時(shí)，我們經(jīng)歷了多個(gè)挑戰(zhàn)，這使得 shuffle 操作成為我們基礎(chǔ)設(shè)施中的擴(kuò)展和性能瓶頸。

可靠性問題

第一個(gè)挑戰(zhàn)是可靠性問題。在我們的生產(chǎn)集群中，由于計(jì)算節(jié)點(diǎn)數(shù)據(jù)量大和 shuffle 工作負(fù)載的規(guī)模，我們注意到集群高峰時(shí)段的 shuffle 服務(wù)可用性問題。這可能會導(dǎo)致 shuffle fetch 失敗，從而導(dǎo)致昂貴的 stage 重試，這可能非常具有破壞性，因?yàn)樗鼈儠?dǎo)致工作流 SLA 違規(guī)和作業(yè)失敗。圖 3 進(jìn)一步說明了這一點(diǎn)，其中顯示了由于 shuffle 導(dǎo)致的每日 Spark stage 失敗的數(shù)量。到 2019 年底，這一趨勢越來越嚴(yán)重，每天有數(shù)百到一千多個(gè) stage 因?yàn)檫@個(gè)失敗。

效率問題

第二個(gè)挑戰(zhàn)是效率問題。在 LinkedIn，我們將 shuffle 文件存儲在 HDD 上。由于 reducer 的 shuffle fetch 請求是隨機(jī)到達(dá)的，因此 shuffle 服務(wù)也會隨機(jī)訪問 shuffle 文件中的數(shù)據(jù)。如果單個(gè) shuffle 塊大小較小，則 shuffle 服務(wù)產(chǎn)生的小隨機(jī)讀取會嚴(yán)重影響磁盤吞吐量，從而延長 shuffle fetch 等待時(shí)間。這也在下圖中進(jìn)行了說明。正如下圖所示，在 2020 年 3 月至 8 月之間，我們生產(chǎn)集群的 10-20% 的計(jì)算資源浪費(fèi)在 shuffle 上，在等待遠(yuǎn)程 shuffle 數(shù)據(jù)時(shí)處于空閑狀態(tài)。

擴(kuò)展性問題

第三個(gè)挑戰(zhàn)是擴(kuò)展問題。由于 external shuffle service 是我們基礎(chǔ)架構(gòu)中的共享服務(wù)，因此一些對 shuffle services 錯(cuò)誤調(diào)優(yōu)的作業(yè)也會影響其他作業(yè)。當(dāng)一個(gè)作業(yè)錯(cuò)誤地配置導(dǎo)致產(chǎn)生許多小的 shuffle blocks 將會給 shuffle 服務(wù)帶來壓力時(shí)，它不僅會給自身帶來性能下降，還會使共享相同 shuffle 服務(wù)的所有相鄰作業(yè)的性能下降。這可能會導(dǎo)致原本正常運(yùn)行的作業(yè)出現(xiàn)不可預(yù)測的運(yùn)行時(shí)延遲，尤其是在集群高峰時(shí)段。

Magnet shuffle service

為了解決這些問題，我們設(shè)計(jì)并實(shí)現(xiàn)了 Magnet，這是一種新穎的基于推送（push-based）的 shuffle 服務(wù)。Magnet 項(xiàng)目在今年早些時(shí)候作為 VLDB 2020 上發(fā)表的工業(yè)跟蹤論文首次向社區(qū)亮相，您可以在此處閱讀我們的 VLDB 論文：Magnet： Push-based Shuffle Service for Large-scale Data Processing。最近，我們正在將 Magnet 回饋給 Apache Spark 社區(qū)。這篇博文的其余部分將介紹 Magnet 背后的高級設(shè)計(jì)及其在生產(chǎn)中的性能，但感興趣的讀者可以到 SPARK-30602 中獲取有關(guān)該工作的更新以及 SPIP 文檔以獲取實(shí)現(xiàn)級別的詳細(xì)信息。

Push-based shuffle

Magnet shuffle 服務(wù)背后的核心思想是基于推送的 shuffle 概念，其中 mapper 生成的 shuffle 塊也被推送到遠(yuǎn)程 shuffle 服務(wù)，以按每個(gè) shuffle 分區(qū)進(jìn)行合并。push-based shuffle 的 shuffle 寫入路徑如下圖所示：

在 map 任務(wù)生成其 shuffle 文件后，它準(zhǔn)備將 shuffle 塊推送到遠(yuǎn)程 ESS。它將 shuffle 文件中的連續(xù)塊組裝成 MB 大小的塊，并將該組數(shù)據(jù)推到相應(yīng)的 ESS。大于特定大小的 Shuffle 塊將被跳過，因此我們不會推送可能來自大型傾斜分區(qū)的塊。map 任務(wù)以一致的方式確定這個(gè)分組和相應(yīng)的 ESS 目的地，從而將屬于同一個(gè) shuffle 分區(qū)的不同 mappers 的塊推到同一 ESS。分組完成后，這些塊的傳輸將移交給專用線程池，然后 map 任務(wù)就完成了。通過這種方式，我們將任務(wù)執(zhí)行線程與塊傳輸線程解耦，在 I/O 密集型數(shù)據(jù)傳輸和 CPU 密集型任務(wù)執(zhí)行之間實(shí)現(xiàn)了更好的并行性。ESS 接受遠(yuǎn)程推送的 shuffle 塊，并將相同分區(qū)的 shuffle 數(shù)據(jù)合并到相應(yīng)的 shuffle 文件中。這是以盡力而為的方式完成的，這并不能保證所有塊都被合并。但是，ESS 確實(shí)保證在合并期間不會發(fā)生數(shù)據(jù)重復(fù)或損壞。

在基于推送的 shuffle 數(shù)據(jù)讀取路徑上，reduce 任務(wù)可以從合并的 shuffle 文件和 map 任務(wù)生成的原始 shuffle 文件中獲取其任務(wù)輸入（如上圖）。ESS 在從合并的 shuffle 文件中讀取時(shí)可以執(zhí)行大的順序 I/O 而不是小的隨機(jī) I/O，從而顯著提高 I/O 效率。利用這一點(diǎn)，reduce 任務(wù)更愿意從合并的 shuffle 文件中獲取它們的輸入。由于塊推送/合并過程是盡力而為的，reduce 任務(wù)可以使用未合并的塊來填充合并的 shuffle 文件中的任何漏洞。如果合并的 shuffle 文件變得不可用，他們甚至可以完全回退到獲取未合并的塊。Magnet 的高效磁盤 I/O 模式進(jìn)一步為構(gòu)建高性能 Spark 集群提供了更大的靈活性，因?yàn)樗僖蕾?SSD 來實(shí)現(xiàn)良好的 shuffle 性能。

Spark driver 負(fù)責(zé)協(xié)調(diào) map 和 reduce 任務(wù)中基于推送的 shuffle。在 shuffle 寫入路徑上，Spark driver 為給定 shuffle 的 map 任務(wù)確定要使用的 ESS 列表。這個(gè) ESS 列表作為任務(wù)上下文的一部分發(fā)送到 Spark executors，這使 map 任務(wù)能夠在塊組和遠(yuǎn)程 ESS 目的地之間提出上述一致的映射。Spark 驅(qū)動程序進(jìn)一步協(xié)調(diào) map 和 reduce 階段之間的轉(zhuǎn)換。一旦所有的 map 任務(wù)都完成了，Spark 驅(qū)動程序會等待一段可配置的時(shí)間，然后通知所有選擇的 ESS 進(jìn)行這次 shuffle 以完成合并操作。當(dāng) ESS 收到終結(jié)請求時(shí)，它停止接受來自給定 shuffle 的任何新塊。它還向驅(qū)動程序響應(yīng)每個(gè)最終 shuffle 分區(qū)的元數(shù)據(jù)列表，其中包括有關(guān)合并的 shuffle 文件的位置和大小信息以及指示哪些塊已合并的位圖。一旦 Spark 驅(qū)動程序從所有 ESS 接收到此類元數(shù)據(jù)，它就會啟動 reduce 階段。此時(shí)，Spark 驅(qū)動程序擁有 shuffle 數(shù)據(jù)位置的完整視圖，現(xiàn)在在合并的 shuffle 文件和原始 shuffle 文件之間進(jìn)行了 2 次復(fù)制。Spark 驅(qū)動程序利用此信息來協(xié)調(diào) reduce 任務(wù)的輸入位置。此外，合并的 shuffle 文件的位置為 reduce 任務(wù)創(chuàng)建了自然的位置偏好。Spark 驅(qū)動程序利用該信息可以在存儲了 shuffle 信息的機(jī)器上啟動 reduce 任務(wù)，如下圖所示：

push-based shuffle 的優(yōu)勢

Push-based shuffle 為 Spark shuffle 帶來了幾個(gè)關(guān)鍵好處。

提高磁盤 I/O 效率

使用 push-based shuffle，shuffle 服務(wù)在訪問 shuffle 文件中的 shuffle 數(shù)據(jù)時(shí)，從小的隨機(jī)讀取切換到大的順序讀取，顯著提高了磁盤 I/O 效率，特別是對于基于 HDD 的 shuffle 存儲。在 shuffle 寫路徑上，即使對小塊進(jìn)行兩次 shuffle 數(shù)據(jù)寫入，整體的 I/O 效率還是有提升的。這是因?yàn)樾〉碾S機(jī)寫入可以從多個(gè)級別的緩存中受益，例如操作系統(tǒng)頁面緩存和磁盤緩沖區(qū)。因此，小隨機(jī)寫入可以實(shí)現(xiàn)比小隨機(jī)讀取高得多的吞吐量。改進(jìn)的磁盤 I/O 效率的效果反映在本博文后面顯示的性能數(shù)據(jù)中。關(guān)于 I/O 效率提升的更詳細(xì)分析包含在我們的 VLDB 論文中。

緩解 shuffle 的可靠性/可擴(kuò)展性問題

Spark 原生 shuffle 操作要成功，需要每個(gè) reduce 任務(wù)成功地從所有 map 任務(wù)中獲取每個(gè)相應(yīng)的 shuffle 塊，這在擁有數(shù)千個(gè)節(jié)點(diǎn)的繁忙集群中通常無法滿足。Magnet 通過多種方式實(shí)現(xiàn)了更好的 shuffle 可靠性：

?Magnet 采用盡力而為的方法來合并塊。塊推送/合并過程中的失敗不會影響 shuffle 的過程。

?通過基于推送的 shuffle，Magnet 有效地生成了 shuffle 中間數(shù)據(jù)的第二個(gè)副本。只有在無法從原始 shuffle 文件或合并的 shuffle 文件中獲取 shuffle 塊時(shí)，才會發(fā)生 shuffle fetch 失敗。

通過 reduce 任務(wù)的位置感知調(diào)度，它們通常在合并后的 shuffle 文件所在的機(jī)器上啟動，這允許它們繞過 ESS 讀取 shuffle 數(shù)據(jù)。這使得 reduce 任務(wù)對 ESS 可用性或性能問題更具彈性，從而緩解前面提到的可擴(kuò)展性問題。

在塊推送過程中處理 stragglers

在 Spark 的普通 shuffle 操作中，由于多個(gè)任務(wù)通常是并發(fā)運(yùn)行的，任務(wù)中的掉隊(duì)者（即一些任務(wù)運(yùn)行速度明顯慢于其他任務(wù)）的影響可以被其他任務(wù)隱藏。使用基于推送的 shuffle，如果在塊推送操作中有任何落后者，它可能會暫停執(zhí)行很長時(shí)間的作業(yè)。這是因?yàn)閴K推送操作介于 shuffle map 和 reduce 階段之間。當(dāng)存在落后者時(shí)，可能根本沒有任務(wù)在運(yùn)行。然而，通過提前終止技術(shù)，Magnet 可以在塊推送過程中有效地處理掉隊(duì)者。Magnet 不是等待 push 過程完全完成，而是限制它在 shuffle map 和 reduce 階段之間等待的時(shí)間。Magnet 的盡力而為的特性使其能夠容忍由于提前終止而未合并的塊。如下圖所示：

與 Spark 原生集成

Magnet 與 Spark 原生集成，這帶來了多種好處：

?Magnet 不依賴于其他外部系統(tǒng)。這有助于簡化 Magnet shuffle 服務(wù)的部署、監(jiān)控和生產(chǎn)。

?通過與 Spark 的 shuffle 系統(tǒng)的原生集成，Magnet shuffle 服務(wù)中的元數(shù)據(jù)可以暴露給 Spark 驅(qū)動程序。這使 Spark 驅(qū)動程序能夠?qū)崿F(xiàn)更好的性能（通過任務(wù)的位置感知調(diào)度）和更好的容錯(cuò)（通過回退到原始 shuffle 塊）。

?Magnet 與現(xiàn)有的 Spark 功能（例如自適應(yīng)查詢執(zhí)行）配合得很好。Spark AQE 的承諾之一是能夠動態(tài)優(yōu)化 skew join，這也需要對 shuffle 進(jìn)行特殊處理。Spark AQE 會在多個(gè) reducer 任務(wù)之間劃分一個(gè)傾斜的 shuffle 分區(qū)，每個(gè)任務(wù)只從 mapper 任務(wù)的一個(gè)子范圍中獲取 shuffle 塊。由于合并后的 shuffle 文件不再保持每個(gè)單獨(dú)的 shuffle 塊的原始邊界，因此無法按照 Spark AQE 要求的方式劃分合并后的 shuffle 文件。由于 Magnet 保留原始 shuffle 文件和合并后的 shuffle 文件，因此它可以委托 AQE 處理偏斜分區(qū)，同時(shí)優(yōu)化非偏斜分區(qū)的 shuffle 操作。

性能對比

我們在 LinkedIn 上使用真實(shí)的生產(chǎn)作業(yè)評估了 Magnet 的性能，我們看到了非常不錯(cuò)的結(jié)果。在下表中，我們展示了運(yùn)行一個(gè) ML 特征生成作業(yè)的性能結(jié)果，該作業(yè)具有數(shù)十個(gè) shuffle 階段和接近 2 TB 的 shuffle 數(shù)據(jù)。與 Spark 中的原生 shuffle 相比，Magnet 取得了非常好的性能結(jié)果。請注意，Magnet 將 shuffle fetch 等待時(shí)間減少了 98%。這可以通過 Magnet 的高效隨機(jī)磁盤 I/O 和減少任務(wù)的位置感知調(diào)度來實(shí)現(xiàn)。此外，這個(gè)作業(yè)并不完全使用 Spark SQL，因?yàn)樗旌鲜褂昧?Spark SQL 和非 SQL 代碼組成其計(jì)算邏輯。在優(yōu)化 shuffle 操作時(shí)，Magnet 只承擔(dān)很少的工作本身。

Total shuffle fetch wait time （min）Total executor task runtime （min）End-to-end job runtime （min）

Vanilla Spark shuffle206365077142

Magnet shuffle445 （-98%）29928 （-41%）31 （-26%）

我們還在 LinkedIn 引入了含有大量 shuffle 的作業(yè)。我們的一個(gè)生產(chǎn)集群中估計(jì)有 15% 的 shuffle 工作負(fù)載已遷移到 Magnet。在這些作業(yè)中，我們看到 shuffle fetch 等待時(shí)間、任務(wù)總運(yùn)行時(shí)間和作業(yè)端到端運(yùn)行時(shí)間也相應(yīng)的減少。如下圖所示，啟用 Magnet 的 Spark 作業(yè)平均減少了 3-4 倍的隨機(jī)提取等待時(shí)間。此外，我們已經(jīng)看到本地訪問的 shuffle 數(shù)據(jù)量增加了大約 10 倍，這表明基于推送的 shuffle 大大改善了數(shù)據(jù)局部性。最后，我們已經(jīng)看到作業(yè)運(yùn)行時(shí)間在集群高峰時(shí)段變得更加穩(wěn)定。隨著我們加入更多的作業(yè)，Magnet 將更多的 shuffle 工作負(fù)載轉(zhuǎn)換為優(yōu)化路徑，從而減輕了 shuffle 服務(wù)的壓力，并為集群帶來更多好處。另一方面，Magnet 可能會將 shuffle 臨時(shí)存儲需求加倍。我們正在通過為 shuffle 文件切換到 zstd 壓縮編解碼器來緩解這種情況，與默認(rèn)壓縮編解碼器相比，這有可能將 shuffle 文件大小減少 50%。

結(jié)論和未來工作

在這篇博文中，我們介紹了 Magnet shuffle 服務(wù)，這是 Apache Spark 的下一代 shuffle 架構(gòu)。Magnet 提高了 Spark 中 shuffle 操作的整體效率、可靠性和可擴(kuò)展性。最近，我們也看到了業(yè)界針對 shuffle 過程提出的其他解決方案，比如 Cosco、Riffle、Zeus 和 Sailfish。我們在 VLDB 論文中對 Magnet 和其他這些解決方案進(jìn)行了比較，尤其是 Cosco、Riffle 和 Sailfish。

未來，我們還將考慮在其他部署環(huán)境和計(jì)算引擎中提供基于 Magnet 推送的 shuffle。我們當(dāng)前的集群部署模式為存儲和計(jì)算是在一起的。隨著 LinkedIn 正在向 Azure 遷移，我們也在評估計(jì)算和存儲分離的集群中基于推送的 shuffle 的方法。此外，我們目前基于推送的 shuffle 的設(shè)計(jì)主要針對批處理引擎，我們也在考慮它對流引擎的適用性。

感謝

需要一個(gè)專門的團(tuán)隊(duì)才能將 Magnet 規(guī)模的項(xiàng)目帶來曙光。除了 Min Shen、Ye Zhou 和 Chandni Singh 的努力外，該項(xiàng)目還得到了 Venkata Krishnan Sowrirajan 和 Mridul Muralidharan 的重大貢獻(xiàn)。Erik Krogen、Ron Hu、Minchu Yang 和 Zoe Lin 為 Magnet 的生產(chǎn)部署和可觀察性改進(jìn)做出了貢獻(xiàn)。特別感謝 Yuval Degani 構(gòu)建的 GridBench——這個(gè)工具使得理解各種因素對作業(yè)運(yùn)行時(shí)的影響變得非常容易。特別感謝我們的合作伙伴團(tuán)隊(duì)，尤其是 Jan Bob 和 Qun Li 的團(tuán)隊(duì)，他們是 Magnet 的早期使用者。

像 Magnet 這樣的大型基礎(chǔ)設(shè)施工作需要管理層做出重大而持續(xù)的支持。Sunitha Beeram、Zhe Zhang、Vasanth Rajamani、Eric Baldeschwieler、Kapil Surlakar 和 Igor Perisic：感謝您的堅(jiān)定支持和指導(dǎo)。Magnet 的設(shè)計(jì)也得益于與 Sriram Rao 和 Shirshanka Das 的評論和深入討論。

Magnet 得到 Apache Spark 社區(qū)的大力支持。感謝與 Databricks 的合作以及來自眾多社區(qū)成員的評論。

責(zé)任編輯：haq