作者 |  一錘、明濟

阿里云EMR自2020年推出Remote Shuffle Service(RSS)以來，幫助了諸多客戶解決Spark作業(yè)的性能、穩(wěn)定性問題，并使得存算分離架構(gòu)得以實施。為了更方便大家使用和擴展，RSS在2022年初開源，歡迎各路開發(fā)者共建。RSS的整體架構(gòu)請參考[1]，本文將介紹RSS最新的兩個重要功能：支持Adaptive Query Execution(AQE)，以及流控。

一、RSS支持AQE

1.AQE簡介

自適應(yīng)執(zhí)行(Adaptive Query Execution， AQE)是Spark3的重要功能[2]，通過收集運行時Stats，來動態(tài)調(diào)整后續(xù)的執(zhí)行計劃，從而解決由于Optimizer無法準(zhǔn)確預(yù)估Stats導(dǎo)致生成的執(zhí)行計劃不夠好的問題。AQE主要有三個優(yōu)化場景: Partition合并(Partition Coalescing), Join策略切換(Switch Join Strategy)，以及傾斜Join優(yōu)化(Optimize Skew Join)。這三個場景都對Shuffle框架的能力提出了新的需求。

Partition合并

Partition合并的目的是盡量讓reducer處理的數(shù)據(jù)量適中且均勻，做法是首先Mapper按較多的Partition數(shù)目進(jìn)行Shuffle Write，AQE框架統(tǒng)計每個Partition的Size，若連續(xù)多個Partition的數(shù)據(jù)量都比較小，則將這些Partition合并成一個，交由一個Reducer去處理。過程如下所示。

由上圖可知，優(yōu)化后的Reducer2需讀取原屬于Reducer2-4的數(shù)據(jù)，對Shuffle框架的需求是ShuffleReader需要支持范圍Partition:

def getReader[K, C](
    handle: ShuffleHandle,
    startPartition: Int,
    endPartition: Int,
    context: TaskContext): ShuffleReader[K, C]

Join策略切換

Join策略切換的目的是修正由于Stats預(yù)估不準(zhǔn)導(dǎo)致Optimizer把本應(yīng)做的Broadcast Join錯誤的選擇了SortMerge Join或ShuffleHash Join。具體而言，在Join的兩張表做完Shuffle Write之后，AQE框架統(tǒng)計了實際大小，若發(fā)現(xiàn)小表符合Broadcast Join的條件，則將小表Broadcast出去，跟大表的本地Shuffle數(shù)據(jù)做Join。流程如下：

Join策略切換有兩個優(yōu)化：1. 改寫成Broadcast Join; 2. 大表的數(shù)據(jù)通過LocalShuffleReader直讀本地。其中第2點對Shuffle框架提的新需求是支持Local Read。

傾斜Join優(yōu)化

傾斜Join優(yōu)化的目的是讓傾斜的Partition由更多的Reducer去處理，從而避免長尾。具體而言，在Shuffle Write結(jié)束之后，AQE框架統(tǒng)計每個Partition的Size，接著根據(jù)特定規(guī)則判斷是否存在傾斜，若存在，則把該Partition分裂成多個Split，每個Split跟另外一張表的對應(yīng)Partition做Join。如下所示。

Partiton分裂的做法是按照MapId的順序累加他們Shuffle Output的Size，累加值超過閾值時觸發(fā)分裂。對Shuffle框架的新需求是ShuffleReader要能支持范圍MapId。綜合Partition合并優(yōu)化對范圍Partition的需求，ShuffleReader的接口演化為:

def getReader[K, C](
    handle: ShuffleHandle,
    startMapIndex: Int,
    endMapIndex: Int,
    startPartition: Int,
    endPartition: Int,
    context: TaskContext,
    metrics: ShuffleReadMetricsReporter): ShuffleReader[K, C]

2.RSS架構(gòu)回顧

RSS的核心設(shè)計是Push Shuffle + Partition數(shù)據(jù)聚合，即不同的Mapper把屬于同一個Partition的數(shù)據(jù)推給同一個Worker做聚合，Reducer直讀聚合后的文件。如下圖所示。

在核心設(shè)計之外，RSS還實現(xiàn)了多副本，全鏈路容錯，Master HA，磁盤容錯，自適應(yīng)Pusher，滾動升級等特性，詳見[1]。

3.RSS支持Partition合并

Partition合并對Shuffle框架的需求是支持范圍Partition，在RSS中每個Partition對應(yīng)著一個文件，因此天然支持，如下圖所示。

4.RSS支持Join策略切換

Join策略切換對Shuffle框架的需求是能夠支持LocalShuffleReader。由于RSS的Remote屬性，數(shù)據(jù)存放在RSS集群，僅當(dāng)RSS和計算集群混部的場景下才會存在在本地，因此暫不支持Local Read(將來會優(yōu)化混部場景并加以支持)。需要注意的是，盡管不支持Local Read，但并不影響Join的改寫，RSS支持Join改寫優(yōu)化如下圖所示。

5.RSS支持Join傾斜優(yōu)化

在AQE的三個場景中，RSS支持Join傾斜優(yōu)化是最為困難的一點。RSS的核心設(shè)計是Partition數(shù)據(jù)聚合，目的是把Shuffle Read的隨機讀轉(zhuǎn)變?yōu)轫樞蜃x，從而提升性能和穩(wěn)定性。多個Mapper同時推送給RSS Worker，RSS在內(nèi)存聚合后刷盤，因此Partition文件中來自不同Mapper的數(shù)據(jù)是無序的，如下圖所示。

Join傾斜優(yōu)化需要讀取范圍Map，例如讀Map1-2的數(shù)據(jù)，常規(guī)的做法有兩種：

• 讀取完整文件，并丟棄范圍之外的數(shù)據(jù)。
• 引入索引文件，記錄每個Block的位置及所屬MapId，僅讀取范圍內(nèi)的數(shù)據(jù)。

這兩種做法的問題顯而易見。方法1會導(dǎo)致大量冗余的磁盤讀;方法2本質(zhì)上回退成了隨機讀，喪失了RSS最核心的優(yōu)勢，并且創(chuàng)建索引文件成為通用的Overhead，即使是針對非傾斜的數(shù)據(jù)(Shuffle Write過程中難以準(zhǔn)確預(yù)測是否存在傾斜)。

為了解決以上兩個問題，我們提出了新的設(shè)計：主動Split + Sort On Read。

主動Split

傾斜的Partition大概率Size非常大，極端情況會直接打爆磁盤，即使在非傾斜場景出現(xiàn)大Partition的幾率依然不小。因此，從磁盤負(fù)載均衡的角度，監(jiān)控Partition文件的Size并做主動Split(默認(rèn)閾值256m)是非常必要的。

Split發(fā)生時，RSS會為當(dāng)前Partition重新分配一對Worker(主副本)，后續(xù)數(shù)據(jù)將推給新的Worker。為了避免Split對正在運行的Mapper產(chǎn)生影響，我們提出了Soft Split的方法，即當(dāng)觸發(fā)Split時，RSS異步去準(zhǔn)備新的Worker，Ready之后去熱更新Mapper的PartitionLocation信息，因此不會對Mapper的PushData產(chǎn)生任何干擾。整體流程如下圖所示。

Sort On Read

為了避免隨機讀的問題，RSS采用了Sort On Read的策略。具體而言，F(xiàn)ile Split的首次Range讀會觸發(fā)排序(非Range讀不會觸發(fā))，排好序的文件連同其位置索引寫回磁盤。后續(xù)的Range讀即可保證是順序讀取。如下圖所示。

為了避免多個Sub-Reducer等待同一個File Split的排序，我們打散了各個Sub-Reducer讀取Split的順序，如下圖所示。

Sort優(yōu)化

Sort On Read可以有效避免冗余讀和隨機讀，但需要對Split File(256m)做排序，本節(jié)討論排序的實現(xiàn)及開銷。文件排序包括3個步驟：讀文件，對MapId做排序，寫文件。RSS的Block默認(rèn)256k，Block的數(shù)量大概是1000，因此排序的過程非常快，主要開銷在文件讀寫。整個排序過程大致有三種方案：

• 預(yù)先分配文件大小的內(nèi)存，文件整體讀入，解析并排序MapId，按MapId順序把Block寫回磁盤。
• 不分配內(nèi)存，Seek到每個Block的位置，解析并排序MapId，按MapId順序把原文件的Block transferTo新文件。
• 分配小塊內(nèi)存(如256k)，順序讀完整個文件并解析和排序MapId，按MapId順序把原文件的Block transferTo新文件。

從IO的視角，乍看之下，方案1通過使用足量內(nèi)存，不存在順序讀寫;方案2存在隨機讀和隨機寫;方案3存在隨機寫;直觀上方案1性能更好。然而，由于PageCache的存在，方案3在寫文件時原文件大概率緩存在PageCache中，因此實測下來方案3的性能更好，如下圖所示。

同時方案3無需占用進(jìn)程額外內(nèi)存，故RSS采用方案3的算法。我們同時還測試了Sort On Read跟上述的不排序、僅做索引的隨機讀方法的對比，如下圖所示。

整體流程

RSS支持Join傾斜優(yōu)化的整體流程如下圖所示。

二、RSS流控

流控的主要目的是防止RSS Worker內(nèi)存被打爆。流控通常有兩種方式：

• Client在每次PushData前先向Worker預(yù)留內(nèi)存，預(yù)留成功才觸發(fā)Push。
• Worker端反壓。

由于PushData是非常高頻且性能關(guān)鍵的操作，若每次推送都額外進(jìn)行一次RPC交互，則開銷太大，因此我們采用了反壓的策略。以Worker的視角，流入數(shù)據(jù)有兩個源：

• Client推送的數(shù)據(jù)
• 主副本發(fā)送的數(shù)據(jù)

如下圖所示，Worker2既接收來自Mapper推送的Partition3的數(shù)據(jù)，也接收Worker1發(fā)送的Partition1的副本數(shù)據(jù)，同時會把Partition3的數(shù)據(jù)發(fā)給對應(yīng)的從副本。

其中，來自Mapper推送的數(shù)據(jù)，當(dāng)且僅當(dāng)同時滿足以下條件時才會釋放內(nèi)存：

• Replication執(zhí)行成功
• 數(shù)據(jù)寫盤成功

來自主副本推送的數(shù)據(jù)，當(dāng)且僅當(dāng)滿足以下條件時才會釋放內(nèi)存：

• 數(shù)據(jù)寫盤成功

我們在設(shè)計流控策略時，不僅要考慮限流(降低流入的數(shù)據(jù))，更要考慮泄流(內(nèi)存能及時釋放)。具體而言，高水位我們定義了兩檔內(nèi)存閾值(分別對應(yīng)85%和95%內(nèi)存使用)，低水位只有一檔(50%內(nèi)存使用)。達(dá)到高水位一檔閾值時，觸發(fā)流控，暫停接收Mapper推送的數(shù)據(jù)，同時強制刷盤，從而達(dá)到泄流的目標(biāo)。僅限制來自Mapper的流入并不能控制來自主副本的流量，因此我們定義了高水位第二檔，達(dá)到此閾值時將同時暫停接收主副本發(fā)送的數(shù)據(jù)。當(dāng)水位低于低水位后，恢復(fù)正常狀態(tài)。整體流程如下圖所示。

三、性能測試

我們對比了RSS和原生的External Shufle Service(ESS)在Spark3.2.0開啟AQE的性能。RSS采用混部的方式，沒有額外占用任何機器資源。此外，RSS所使用的內(nèi)存為8g，僅占機器內(nèi)存的2.3%(機器內(nèi)存352g)。具體環(huán)境如下。

1.測試環(huán)境

硬件：

header 機器組 1x ecs.g5.4xlargeworker 機器組 8x ecs.d2c.24xlarge，96 CPU，352 GB，12x 3700GB HDD。

Spark AQE相關(guān)配置:

spark.sql.adaptive.enabled true
spark.sql.adaptive.coalescePartitions.enabled true
spark.sql.adaptive.coalescePartitions.initialPartitionNum 1000
spark.sql.adaptive.skewJoin.enabled true
spark.sql.adaptive.localShuffleReader.enabled false

RSS相關(guān)配置:

RSS_MASTER_MEMORY=2g
RSS_WORKER_MEMORY=1g
RSS_WORKER_OFFHEAP_MEMORY=7g

2.TPCDS 10T測試集

我們測試了10T的TPCDS，E2E來看，ESS耗時11734s，RSS單副本/兩副本分別耗時8971s/10110s，分別比ESS快了23.5%/13.8%，如下圖所示。我們觀察到RSS開啟兩副本時網(wǎng)絡(luò)帶寬達(dá)到上限，這也是兩副本比單副本低的主要因素。

具體每個Query的時間對比如下:

相關(guān)鏈接

github地址：https://github.com/alibaba/RemoteShuffleService

Reference

[1]阿里云EMR Remote Shuffle Service在小米的實踐，以及開源. https://developer.aliyun.com/article/857757

[2]Adaptive Query Execution: Speeding Up Spark SQL at Runtime. https://databricks.com/blog/2020/05/29/adaptive-query-execution-speeding-up-spark-sql-at-runtime.html