作者 | 一錘、明濟、紫槿來源 | 阿裡技術公衆号

阿裡雲EMR自2020年推出Remote Shuffle Service(RSS)以來，幫助了諸多客戶解決Spark作業的性能、穩定性問題，并使得存算分離架構得以實施，與此同時RSS也在跟合作方小米的共建下不斷演進。本文将介紹RSS的最新架構，在小米的實踐，以及開源。

Shuffle是大數據計算中最為重要的算子。首先，覆蓋率高，超過50%的作業都包含至少一個Shuffle[2]。其次，資源消耗大，阿裡内部平台Shuffle的CPU占比超過20%，LinkedIn内部Shuffle Read導緻的資源浪費高達15%[1]，單Shuffle數據量超100T[2]。第三，不穩定，硬件資源的穩定性CPU>内存>磁盤≈網絡，而Shuffle的資源消耗是倒序。OutOfMemory和Fetch Failure可能是Spark作業最常見的兩種錯誤，前者可以通過調參解決，而後者需要系統性重構Shuffle。

傳統Shuffle如下圖所示，Mapper把Shuffle數據按PartitionId排序寫盤後交給External Shuffle Service(ESS)管理，Reducer從每個Mapper Output中讀取屬于自己的Block。

傳統Shuffle存在以下問題。

• 本地盤依賴限制了存算分離。存算分離是近年來興起的新型架構，它解耦了計算和存儲，可以更靈活地做機型設計：計算節點強CPU弱磁盤，存儲節點強磁盤強網絡弱CPU。計算節點無狀态，可根據負載彈性伸縮。存儲端，随着對象存儲(OSS, S3) 數據湖格式(Delta, Iceberg, Hudi) 本地/近地緩存等方案的成熟，可當作容量無限的存儲服務。用戶通過計算彈性 存儲按量付費獲得成本節約。然而，Shuffle對本地盤的依賴限制了存算分離。
• 寫放大。當Mapper Output數據量超過内存時觸發外排，從而引入額外磁盤IO。
• 大量随機讀。Mapper Output屬于某個Reducer的數據量很小，如Output 128M，Reducer并發2000，則每個Reducer隻讀64K，從而導緻大量小粒度随機讀。對于HDD，随機讀性能極差；對于SSD，會快速消耗SSD壽命。
• 高網絡連接數，導緻線程池消耗過多CPU，帶來性能和穩定性問題。
• Shuffle數據單副本，大規模集群場景壞盤/壞節點很普遍，Shuffle數據丢失引發的Stage重算帶來性能和穩定性問題。

針對Shuffle的問題，工業界嘗試了各種方法，近兩年逐漸收斂到Push Shuffle的方案。

Sailfish3最早提出Push Shuffle Partition數據聚合的方法，對大作業有20%-5倍的性能提升。Sailfish魔改了分布式文件系統KFS[4]，不支持多副本。

Goolge BigQuery和Cloud Dataflow5實現了Shuffle跟計算的解耦，采用多層存儲(内存 磁盤)，除此之外沒有披露更多技術細節。

Facebook Riffle2采用了在Mapper端Merge的方法，物理節點上部署的Riffle服務負責把此節點上的Shuffle數據按照PartitionId做Merge，從而一定程度把小粒度的随機讀合并成較大粒度。

Facebook Cosco[6]7采用了Sailfish的方法并做了重設計，保留了Push Shuffle Parititon數據聚合的核心方法，但使用了獨立服務。服務端采用Master-Worker架構，使用内存兩副本，用DFS做持久化。Cosco基本上定義了RSS的标準架構，但受到DFS的拖累，性能上并沒有顯著提升。

Uber Zeus[8]9同樣采用了去中心化的服務架構，但沒有類似etcd的角色維護Worker狀态，因此難以做狀态管理。Zeus通過Client雙推的方式做多副本，采用本地存儲。

Intel RPMP10依靠RDMA和PMEM的新硬件來加速Shuffle，并沒有做數據聚合。

LinkedIn Magnet1融合了本地Shuffle Push Shuffle，其設計哲學是"盡力而為"，Mapper的Output寫完本地後，Push線程會把數據推給遠端的ESS做聚合，且不保證所有數據都會聚合。受益于本地Shuffle，Magnet在容錯和AE的支持上的表現更好(直接FallbACK到傳統Shuffle)。Magnet的局限包括依賴本地盤，不支持存算分離；數據合并依賴ESS，對NodeManager造成額外壓力；Shuffle Write同時寫本地和遠端，性能達不到最優。Magnet方案已經被Apache Spark接納，成為默認的開源方案。

FireStorm11混合了Cosco和Zeus的設計，服務端采用Master-Worker架構，通過Client多寫實現多副本。FireStorm使用了本地盤 對象存儲的多層存儲，采用較大的PushBlock(默認3M)。FireStorm在存儲端保留了PushBlock的元信息，并記錄在索引文件中。FireStorm的Client緩存數據的内存由Spark MemoryManager進行管理，并通過細顆粒度的内存分配(默認3K)來盡量避免内存浪費。

從上述描述可知，當前的方案基本收斂到Push Shuffle，但在一些關鍵設計上的選擇各家不盡相同，主要體現在:

1. 集成到Spark内部還是獨立服務。
2. RSS服務側架構，選項包括：Master-Worker，含輕量級狀态管理的去中心化，完全去中心化。
3. Shuffle數據的存儲，選項包括：内存，本地盤，DFS，對象存儲。
4. 多副本的實現，選項包括：Client多推，服務端做Replication。

阿裡雲RSS12由2020年推出，核心設計參考了Sailfish和Cosco，并且在架構和實現層面做了改良，下文将詳細介紹。

針對上一節的關鍵設計，阿裡雲RSS的選擇如下：

1. 獨立服務。考慮到将RSS集成到Spark内部無法滿足存算分離架構，阿裡雲RSS将作為獨立服務提供Shuffle服務。
2. Master-Worker架構。通過Master節點做服務狀态管理非常必要，基于etcd的狀态狀态管理能力受限。
3. 多種存儲方式。目前支持本地盤/DFS等存儲方式，主打本地盤，将來會往分層存儲方向發展。
4. 服務端做Replication。Client多推會額外消耗計算節點的網絡和計算資源，在獨立部署或者服務化的場景下對計算集群不友好。

下圖展示了阿裡雲RSS的關鍵架構，包含Client(RSS Client, Meta Service)，Master(Resource Manager)和Worker三個角色。Shuffle的過程如下:

1. Mapper在首次PushData時請求Master分配Worker資源，Worker記錄自己所需要服務的Partition列表。
2. Mapper把Shuffle數據緩存到内存，超過阈值時觸發Push。
3. 隸屬同個Partition的數據被Push到同一個Worker做合并，主Worker内存接收到數據後立即向從Worker發起Replication，數據達成内存兩副本後即向Client發送ACK，Flusher後台線程負責刷盤。
4. Mapper Stage運行結束，MetaService向Worker發起CommitFiles命令，把殘留在内存的數據全部刷盤并返回文件列表。
5. Reducer從對應的文件列表中讀取Shuffle數據。

點擊鍊接查看原文阿裡雲EMR Remote Shuffle Service在小米的實踐，關注公衆号【阿裡技術】獲取更多福利！

更多精彩资讯请关注tft每日頭條，我们将持续为您更新最新资讯!