1. 背景介紹

2018年B站基于Hadoop開(kāi)始搭建離線計(jì)算服務(wù)，計(jì)算集群規(guī)模從最初的兩百臺(tái)到發(fā)展到目前近萬(wàn)臺(tái)，從單機(jī)房發(fā)展到多機(jī)房。我們先后在生產(chǎn)上大規(guī)模的使用了 Hive、Spark、Presto 作為離線計(jì)算引擎，其中 Hive 和 Spark 部署在 Yarn 上，具體的架構(gòu)如下，目前每天有約20w的離線批作業(yè)運(yùn)行在 Spark 和 Hive 上，下面介紹下我們做了哪些工作來(lái)確保這些作業(yè)的高效與穩(wěn)定。

2. 從Hive到Spark

21年初的時(shí)候Hive還是B站主要的離線計(jì)算引擎，80%以上的離線作業(yè)使用 Hive 執(zhí)行，Spark2.4作業(yè)占比接近20%，集群資源的使用率長(zhǎng)期維持在80%以上。21年3月 Spark3.1 發(fā)布，相較于 Spark2.4 性能有了較大的提升，我們開(kāi)始推動(dòng)Spark3.1 在B站的落地，同時(shí)將 Hive-SQL 整體遷移至 Spark-SQL。

在B站，離線計(jì)算的調(diào)度已經(jīng)完成了收口，80%以上的作業(yè)來(lái)自于自建的 BSK 調(diào)度平臺(tái)，其余的作業(yè)基本也都是 airflow 提交上來(lái)的，只有少量的任務(wù)來(lái)自散落的開(kāi)發(fā)機(jī)。在推動(dòng) Hive 升級(jí) Spark 時(shí)只要將調(diào)度平臺(tái)的作業(yè)完成遷移就可以覆蓋90%以上的作業(yè)。起步階段我們進(jìn)行了少量的人工遷移，對(duì)用戶 SQL 進(jìn)行了簡(jiǎn)單改寫，修改了輸入輸出表后由兩個(gè)引擎執(zhí)行，開(kāi)發(fā)了一個(gè)結(jié)果對(duì)比的工具，通過(guò)對(duì)雙跑結(jié)果分析保障遷移效果。基于這個(gè)操作鏈路我們自研了一個(gè)自動(dòng)遷移工具，減少人工失誤和人力壓力。

2.1 語(yǔ)句轉(zhuǎn)換

我們重寫了 SparkSqlParser，將從調(diào)度系統(tǒng)中收集到的 SQL 進(jìn)行輸入輸出表的替換，避免對(duì)生產(chǎn)環(huán)境的影響。調(diào)度平臺(tái)進(jìn)行作業(yè)調(diào)度時(shí)以 DAG 為單位，一個(gè)調(diào)度任務(wù)里面可能存在多條 SQL，這些 SQL的輸入輸出表間存在依賴關(guān)系，為了保證雙跑盡可能的模擬生產(chǎn)表現(xiàn)，對(duì)一個(gè) DAG 里面的多個(gè)調(diào)度作業(yè)進(jìn)行輸入輸出表替換時(shí)進(jìn)行整體替換，保證了相互間依賴。對(duì)于 Select語(yǔ)句因?yàn)楸旧頉](méi)有輸出表，需要將 Select 語(yǔ)句轉(zhuǎn)換為 CTAS 語(yǔ)句，這樣就能將執(zhí)行結(jié)果落地進(jìn)行對(duì)比，需要注意的是轉(zhuǎn)換過(guò)程中要將列名進(jìn)行編碼防止中文列導(dǎo)致的建表失敗。當(dāng)遷移工具識(shí)別出 SQL 語(yǔ)句為 DDL 語(yǔ)句，如果不是 CTAS 這種需要消耗計(jì)算資源的就直接跳過(guò)對(duì)比，同時(shí)對(duì)該語(yǔ)句進(jìn)行標(biāo)記，保證交由 Hive 執(zhí)行，防止意外的元信息修改。

2.2 結(jié)果對(duì)比

雙跑輸出結(jié)果的對(duì)比是保證數(shù)據(jù)準(zhǔn)確性的關(guān)鍵。首先對(duì)兩個(gè)結(jié)果表的 Schema 進(jìn)行對(duì)比，這個(gè)通過(guò)調(diào)用 DESC 語(yǔ)法返回結(jié)果對(duì)照就可以完成。對(duì)于 Schema 一致的兩個(gè)表則進(jìn)行下一步操作，兩表全量數(shù)據(jù)對(duì)比，我們?cè)O(shè)計(jì)了一個(gè) SQL 對(duì)數(shù)據(jù)按行進(jìn)行整體對(duì)比，具體的對(duì)比思路如圖：

第一步將兩表按所有列(這里是 name 和 num 字段)進(jìn)行 GROUP BY，第二步 UNION ALL 兩表數(shù)據(jù)，第三步再按所有列(這里是 name, num 和 cnt 字段) GROUP BY 一次產(chǎn)生最終表，在最終表中 cnts 值為2的行表示這行數(shù)據(jù)在兩表中都有且重復(fù)值一致，對(duì)于值非2的數(shù)據(jù)就是差異行了。從上圖的例子來(lái)說(shuō)差異行 Jack|1|2|1 表示 Jack|1 這行數(shù)據(jù)數(shù)據(jù)在一個(gè)表中存在兩行，結(jié)合差異行 Jack|1|1|1 來(lái)看其實(shí)就是 Jack|1 這行數(shù)據(jù)一個(gè)表有一行另一個(gè)表有兩行。通過(guò)這個(gè)方式就可以對(duì)雙跑產(chǎn)出的結(jié)果表進(jìn)行一個(gè)全量的對(duì)比。通過(guò)這種結(jié)果對(duì)比方法可以完成大部分雙跑任務(wù)的結(jié)果對(duì)比，但是對(duì)于結(jié)果表中存在 LIST、SET、MAP 這種容器類型的，因?yàn)樵?toString 時(shí)順序是無(wú)法保證的，所以會(huì)被識(shí)別為不一致，此外對(duì)于非穩(wěn)定性的 SQL 如某列數(shù)據(jù)是 random 產(chǎn)生，因?yàn)槊看螆?zhí)行產(chǎn)出的結(jié)果不一致，也會(huì)識(shí)別為對(duì)比失敗，這兩種情況下就需用人工的介入來(lái)分析了。

資源利用率的提升是做引擎升級(jí)的出發(fā)點(diǎn)，除了結(jié)果對(duì)比來(lái)保證數(shù)據(jù)準(zhǔn)確性，我們還做了資源消耗對(duì)比來(lái)保證遷移的收益。對(duì)比系統(tǒng)收集了每個(gè)作業(yè)的執(zhí)行時(shí)間以及消耗的資源，從執(zhí)行時(shí)間、CPU 和內(nèi)存的資源消耗進(jìn)行兩個(gè)引擎執(zhí)行性能的對(duì)比，在執(zhí)行最終遷移前依據(jù)收集的數(shù)據(jù)為用戶提供了遷移的預(yù)期收益，提高了用戶遷移任務(wù)的積極性。從遷移中收集的數(shù)據(jù)來(lái)看 hive 切到 spark 可以減少40%以上的執(zhí)行時(shí)間，同時(shí)整體資源消耗降低30%以上。

2.3 遷移&回滾

遷移系統(tǒng)對(duì)每個(gè)任務(wù)都執(zhí)行了至少3次的雙跑對(duì)比，但依然不能完全消除執(zhí)行遷移的風(fēng)險(xiǎn)，在實(shí)際遷移過(guò)程中的幾次問(wèn)題都是遷移后穩(wěn)定性不符合預(yù)期導(dǎo)致的，因此遷移系統(tǒng)對(duì)于遷移后的任務(wù)增加了監(jiān)控，在一個(gè)任務(wù)遷移后，該任務(wù)的前3次調(diào)度執(zhí)行消耗的時(shí)間、CPU 和內(nèi)存資源將被用來(lái)和遷移前的七次平均執(zhí)行數(shù)據(jù)對(duì)比，如果存在負(fù)優(yōu)化的情況則會(huì)將這個(gè)任務(wù)執(zhí)行引擎進(jìn)行回滾并通知我們介入進(jìn)行進(jìn)一步分析。

3. Spark 在B站的實(shí)踐

3.1 穩(wěn)定性改進(jìn)

3.1.1 小文件問(wèn)題

隨著B(niǎo)站業(yè)務(wù)高速發(fā)展，數(shù)據(jù)量和作業(yè)數(shù)增長(zhǎng)越來(lái)越快，伴隨而來(lái)的小文件數(shù)也快速增長(zhǎng)，小文件太多會(huì)增加 HDFS 元數(shù)據(jù)的壓力，在計(jì)算引擎讀取時(shí)也大大增加了讀請(qǐng)求的數(shù)量降低了讀取效率。為了解決小文件的問(wèn)題，在寫表場(chǎng)景下對(duì) Spark 做了如下兩種改造。

兜底小文件合并：我們修改了數(shù)據(jù)的寫出目錄，引擎計(jì)算先寫到一個(gè)中間目錄，在 FileFormatWriter.write 結(jié)束后 refreshUpdatedPartitions 前，插入了一個(gè)文件合并邏輯，從中間目錄中獲取分區(qū)下文件的平均大小，對(duì)于不存在小文件情況的目錄直接MV到最終目錄，對(duì)于存在小文件的目錄新增一個(gè)讀 RDD coalesce 到一個(gè)合適值寫出后 MV 到最終目錄。

基于 reparation 的小文件合并：可以看到兜底小文件合并方式需要先將數(shù)據(jù)落地到 HDFS，重新讀取后再寫出，這樣做放大了 HDFS寫操作(三副本)，降低了計(jì)算引擎的執(zhí)行性能。而 Spark3的 AQE 特性可以在有 shuffle 的場(chǎng)景下有效解決小文件的問(wèn)題，很多情況下對(duì)于沒(méi)有 shuffle 的場(chǎng)景新增一個(gè) reparation 操作就可以借助 AQE 的能力解決小文件的問(wèn)題。社區(qū) AQE 對(duì)于 reparation 這個(gè) hint 是不會(huì)調(diào)整目標(biāo)分區(qū)數(shù)的，我們新增了一個(gè) rebalance hint，本質(zhì)上和reparation 一樣只是將 AQE 的特性應(yīng)用在了這個(gè)操作上，同時(shí)將 AQE 目標(biāo) size 相關(guān)的屬性和 rebalance 設(shè)置屬性做了隔離方便更好的設(shè)置文件大小而不影響計(jì)算的并行度。rebalance 操作會(huì)在最終寫出前增加一個(gè) shuffle stage，有些情況下沒(méi)有這個(gè) stage 上游輸出就已經(jīng)沒(méi)有小文件了，為此作業(yè)是否增加 rebalance 操作依賴于我們對(duì)任務(wù)的畫像通過(guò) HBO 系統(tǒng)開(kāi)啟。

3.1.2 shuffle 穩(wěn)定性問(wèn)題

Shuffle 穩(wěn)定性直接影響了 Spark 作業(yè)的 SLA，在B站推動(dòng) Spark 升級(jí)過(guò)程中成為用戶顧慮的點(diǎn)。

shuffle 磁盤分級(jí)：B站 Yarn 主集群采用 DataNode 和 NodeManage 混部模式，節(jié)點(diǎn)配置了多塊 HDD 盤和少量 SSD 盤，NM 以 HDD 盤作為計(jì)算盤，由于和 DN 沒(méi)有做到 IO 隔離，DN 和shuffle service 經(jīng)常互相影響，因此我們對(duì)DiskBlockManager 進(jìn)行了改造，優(yōu)先使用 SSD 盤下的目錄作為工作目錄，當(dāng) SSD 盤存儲(chǔ)空間或者 inode 緊張時(shí)則降級(jí)到 Yarn 配置的計(jì)算目錄，借助 SSD 優(yōu)異的隨機(jī) IO 能力，有效的提高的了 shuffle 穩(wěn)定性。

remote shuffle service：push based shuffle 方案可以大量降低磁盤隨機(jī)IO讀請(qǐng)求，如下圖：

通過(guò)中間服務(wù)將同屬一個(gè)分區(qū)的數(shù)據(jù)進(jìn)行歸并，后續(xù) reduce 操作就不需要從上游所有的 Map 節(jié)點(diǎn)拉取數(shù)據(jù)，在 shuffle 上下游 Task 數(shù)量多的情況下會(huì)對(duì)磁盤 IO 壓力指數(shù)放大，生產(chǎn)上 shuffle heavy 的任務(wù)表現(xiàn)很不穩(wěn)定，經(jīng)常出現(xiàn)FetchFailed Exception。B站在推動(dòng) RSS 落地時(shí)選擇了社區(qū)3.2 Push based shuffle 的方案，這個(gè)方案主要的優(yōu)點(diǎn)是對(duì) AQE 支持比較好，缺點(diǎn)是因?yàn)楸镜匾惨獙懸环輸?shù)據(jù)放大了寫。

將數(shù)據(jù)先寫本地后異步的發(fā)送到 driver 維護(hù)的 executor 節(jié)點(diǎn)的 external shuffle 節(jié)點(diǎn)上，后續(xù)生產(chǎn)實(shí)踐中該方案有個(gè)問(wèn)題，就是當(dāng)作業(yè)啟動(dòng)時(shí)通常 driver 維護(hù)的 executor 數(shù)不足以滿足遠(yuǎn)程節(jié)點(diǎn)的選擇，而 SQL 作業(yè)參與計(jì)算的數(shù)據(jù)量通常是隨著過(guò)濾條件層層遞減的，通常 shuffle 數(shù)據(jù)量大的時(shí)候因?yàn)闆](méi)有足夠的節(jié)點(diǎn)會(huì) fall back 到原先的 shuffle 方式，為了解決這個(gè)問(wèn)題，我們新增了 shuffle  service master 節(jié)點(diǎn)，具體調(diào)用流程如下圖，所有的 external shuffle 節(jié)點(diǎn)啟動(dòng)時(shí)都會(huì)注冊(cè)到 shuffle master 節(jié)點(diǎn)上，后續(xù)節(jié)點(diǎn)本身也會(huì)周期性的上報(bào)心跳和節(jié)點(diǎn)繁忙程度，DAGScheduler 后續(xù)請(qǐng)求遠(yuǎn)程節(jié)點(diǎn)都從 shuffle master 申請(qǐng)，這樣不僅解決了冷啟動(dòng)節(jié)點(diǎn)不足的問(wèn)題，在節(jié)點(diǎn)選擇上也考慮了節(jié)點(diǎn)的健康程度。

因?yàn)槭窍嚷浔P后發(fā)送，在 stage 執(zhí)行結(jié)束后會(huì)有一個(gè)等待時(shí)間，這里面會(huì)有個(gè)性能回退的問(wèn)題，對(duì)小任務(wù)不友好，所以在生產(chǎn)應(yīng)用中我們基于任務(wù)畫像系統(tǒng) HBO 自動(dòng)決定任務(wù)是否啟用RSS服務(wù)，目前生產(chǎn)大約7%的大任務(wù)在使用RSS 服務(wù)，這些任務(wù)平均執(zhí)行時(shí)間縮短了25%，穩(wěn)定性有了顯著提升。

目前B站生產(chǎn)中使用該方案基本解決了 shuffle 穩(wěn)定性的問(wèn)題，不過(guò)這套方案依舊需要計(jì)算節(jié)點(diǎn)配置本地 shuffle 盤，在本地落 shuffle 數(shù)據(jù)，無(wú)法支持存算分離的架構(gòu)。后續(xù)我們?cè)?k8s 上會(huì)大規(guī)模上線混部集群，需要盡量不依賴本地磁盤，避免對(duì)在線應(yīng)用的影響，我們也關(guān)注到騰訊和阿里相繼開(kāi)源各自的 RSS 方案，我們也在嘗試在生產(chǎn)中使用純遠(yuǎn)程 shuffle 方案來(lái)滿足 Spark on K8s 的技術(shù)需要。

3.1.3 大結(jié)果集溢寫到磁盤

在adhoc 場(chǎng)景中用戶通常會(huì)拉取大量結(jié)果到 driver 中，造成了大量的內(nèi)存消耗，driver 穩(wěn)定性又直接影響著用戶即席查詢的體驗(yàn)，為此專門優(yōu)化了 executor fetch result 的過(guò)程，在獲取結(jié)果時(shí)會(huì)監(jiān)測(cè) driver 內(nèi)存使用情況，在高內(nèi)存使用下將拉取到的結(jié)果直接寫出到文件中，返回給用戶時(shí)則直接分批從文件中獲取，增加 driver 的穩(wěn)定性。

3.1.4 單 SQL task 并行度、task 數(shù)、執(zhí)行時(shí)間限制

生產(chǎn)上我們按隊(duì)列隔離了用戶的 adhoc 查詢，在實(shí)踐過(guò)程中經(jīng)常性的遇到單個(gè)大作業(yè)直接占用了全部并行度，有些短作業(yè)直接因?yàn)楂@取不到資源導(dǎo)致長(zhǎng)時(shí)間的 pending 的情況，為了解決這種問(wèn)題首先對(duì)單個(gè) SQL 執(zhí)行時(shí)間和總 task 數(shù)進(jìn)行了限制，此外考慮到在 task 調(diào)度時(shí)有資源就會(huì)全部調(diào)度出去，后續(xù) SQL 過(guò)來(lái)就面臨著完全無(wú)資源可用的情況，我們修改了調(diào)度方法對(duì)單個(gè) SQL 參與調(diào)度的 task 數(shù)進(jìn)行了限制，具體的限制數(shù)隨著可用資源進(jìn)行一個(gè)動(dòng)態(tài)變化，在 current executor 數(shù)接近于 max executor 的情況下進(jìn)行嚴(yán)格限制 ，在 current executor 數(shù)明顯少于 max executor 的情況下，提高單 SQL 并行的 task 總數(shù)限制。

3.1.5 危險(xiǎn) join condition 發(fā)現(xiàn)& join 膨脹率檢測(cè)

• 危險(xiǎn) join condition 發(fā)現(xiàn)

在選擇 join 方式的時(shí)候如果是等值 join 則按照 BHJ，SHJ，SMJ 的順序選擇，如果還沒(méi)有選擇出則判斷 Cartesian Join，如果 join 類型是 InnerType 的就使用 Cartesian Join，Cartesian Join 會(huì)產(chǎn)生笛卡爾積比較慢，如果不是 InnerType，則使用 BNLJ，在判斷 BHJ 時(shí)，表的大小就超過(guò)了 broadcast 閾值，因此將表 broadcast 出去可能會(huì)對(duì) driver 內(nèi)存造成壓力，性能比較差甚至可能會(huì) OOM，因此將這兩種 join 類型定義為危險(xiǎn) join。

如果不是等值 join 則只能使用 BNLJ 或者 Cartesian Join，如果在第一次 BNLJ 時(shí)選不出 build side 說(shuō)明兩個(gè)表的大小都超過(guò)了 broadcast 閾值，則使用 Cartesian Join，如果 Join Type 不是 InnerType 則只能使用 BNLJ，因此 Join 策略選擇Cartesian Join 和第二次選擇 BNLJ 時(shí)為危險(xiǎn) join。

• join 膨脹率檢測(cè)

ShareState 中的 statusScheduler 用于收集 Execution 的狀態(tài)和指標(biāo)，這其中的指標(biāo)就是按照 nodes 匯總了各個(gè) task 匯報(bào)上來(lái)的 metrics，我們啟動(dòng)了一個(gè) join 檢測(cè)的線程定時(shí)的監(jiān)控 Join 節(jié)點(diǎn)的 "number of output rows"及 Join 的2個(gè)父節(jié)點(diǎn)的 "number of output rows" 算出該 Join 節(jié)點(diǎn)的膨脹率。

• 傾斜 Key 發(fā)現(xiàn)

數(shù)據(jù)傾斜是 ETL 任務(wù)比較常見(jiàn)的問(wèn)題，以 shuffle 過(guò)程中的傾斜為例，通常有以下幾個(gè)解決方法：增大 shuffle 的分區(qū)數(shù)量從而使數(shù)據(jù)分散到更多的分區(qū)中；修改邏輯，將 shuffle 時(shí)的 key 盡可能打散；單獨(dú)找出產(chǎn)生了極大傾斜的 key，在邏輯中單獨(dú)處理等等。但在進(jìn)行這些處理之前，我們都需要先知道傾斜發(fā)生在 SQL 邏輯的哪個(gè)部分以及發(fā)生傾斜的是哪些 key。為了幫助用戶自助高效的解決數(shù)據(jù)傾斜問(wèn)題，我們實(shí)現(xiàn)了傾斜 key 發(fā)現(xiàn)的功能。以 SortMergeJoin 為例，在 shuffle fetch 階段，首先根據(jù) mapStatuses 計(jì)算出每個(gè) partition size，并根據(jù)一定策略判斷該 task 所處理的 partition 是否傾斜。如果傾斜，則在 join 階段對(duì)數(shù)據(jù)進(jìn)行采樣，找到發(fā)生傾斜的 key，通過(guò) TaskMetric 發(fā)送到 driver 端，driver 端消費(fèi) metric后會(huì)記錄傾斜信息。

上面這些 bad case 在運(yùn)行時(shí)發(fā)現(xiàn)后會(huì)自動(dòng)將信息發(fā)送到我們內(nèi)部作業(yè)診斷平臺(tái)，用戶可以查看并對(duì)語(yǔ)句做優(yōu)化和改進(jìn)。

3.2 性能優(yōu)化

3.2.1 DPP 和 AQE 兼容

spark3.1 的 DPP 和 AQE 存在兼容問(wèn)題，在使用 AQE 后 DPP 的策略就無(wú)法生效，這個(gè)問(wèn)題在3.2得到了修復(fù)，我們將3.2的相關(guān)代碼 backport 回來(lái)，從 TPCDS 測(cè)試上看對(duì)3.1有很明顯的提升。

3.2.2 AQE 支持 ShuffledHashJoin

AQE 通過(guò)對(duì) map 階段收集的指標(biāo)數(shù)據(jù)來(lái)優(yōu)化 Join 方式，對(duì)于存在小表的情況能將 SMJ 優(yōu)化為 BHJ，這個(gè)操作可以顯著的優(yōu)化性能。Spark的 shuffle 策略還有一個(gè)就是 ShuffledHashJoin，該策略性能相對(duì)較好，但內(nèi)存壓力大，在默認(rèn)情況下為了保證任務(wù)的穩(wěn)定性我們將其關(guān)閉，基于 AQE 的思想，在 map 完成后收集 partition size，當(dāng)最大的 partition size 小于定義的值后，通過(guò)新增 DynamicJoin 優(yōu)化策略將 SMJ 優(yōu)化為 SHJ。

3.2.3 Runtime filter

DPP 通過(guò)對(duì)大表直接進(jìn)行 partition 級(jí)別的裁剪，可以大大提高查詢速度，但 DPP 的適用條件也相對(duì)嚴(yán)格，需要大表的分區(qū)列參與 join，但如果大表參與 join 的列為非分區(qū)列則無(wú)法應(yīng)用。我們知道 shuffle 是比較耗時(shí)的操作，shuffle 的數(shù)據(jù)量越大，耗時(shí)越久，而且對(duì)網(wǎng)絡(luò)，機(jī)器 IO 都會(huì)產(chǎn)生比較大的壓力。如果能在大表 shuffle 前根據(jù)非分區(qū)列的 join 列對(duì)其進(jìn)行過(guò)濾，即使無(wú)法像 DPP 一樣直接減少?gòu)拇鎯?chǔ)中讀取的數(shù)據(jù)量，但減小了其參與 shuffle 以及后續(xù)操作的數(shù)據(jù)量，也能獲得比較不錯(cuò)的收益，這就是 runtime filter 的動(dòng)機(jī)，即運(yùn)行時(shí)預(yù)先掃描小表獲取 join 列的值，構(gòu)造 bloom filter 對(duì)大表進(jìn)行過(guò)濾。具體實(shí)現(xiàn)思路和 DPP 基本一致，首先在 SparkOptimizer 新增 DynamicBloomFilterPruning 規(guī)則，邏輯上類似PartitionPruning，符合一系列判斷條件后插入一個(gè)節(jié)點(diǎn) DynamicBloomFilterPruningSubquery。與 DPP 不同的是，如果 join 可以被轉(zhuǎn)化為 BroadcastHashJoin，則不會(huì)應(yīng)用該規(guī)則，因?yàn)樵?BroadcastHashJoin 的情況下對(duì)大表進(jìn)行預(yù)先的過(guò)濾其實(shí)是多余的（非 pushdown 的情況下）。判斷是否加入 filter 節(jié)點(diǎn)的主要邏輯如下，這里以裁剪左表（左右兩側(cè)都為 logicalPlan，為了方便表達(dá)，用左右表指代）為例進(jìn)行說(shuō)明，需要滿足以下條件：

• 右表 rowCount 需要小于左表
• Join 類型支持裁剪左表
  
• 右表 rowCount > 0
  
• 右表 rowCount 小于 spark.sql.optimizer.dynamicBloomFilterJoinPruning.maxBloomFilterEntries，默認(rèn)值為100000000，避免 bloom filter 占用內(nèi)存過(guò)大
  
• 右表中沒(méi)有DynamicBloomFilterPruningSubquery
  
• 右表不是 stream 且存在 SelectivePredicate
  
• 左表(這里的左表是真正的左表或者包含左表的Filter節(jié)點(diǎn))沒(méi)有 SelectivePredicate，因?yàn)槿绻嬖?SelectivePredicate，那么下一步便無(wú)法根據(jù)統(tǒng)計(jì)信息去計(jì)算過(guò)濾收益
  

在 prepare 階段，PlanAdaptiveSubqueries 會(huì)把 DynamicBloomFilterPruningSubquery 節(jié)點(diǎn)替換為 DynamicPruningExpression(InBloomFilterSubqueryExec(_, _, _))，擴(kuò)展了PlanAdaptiveDynamicPruningFilters，支持對(duì)以上節(jié)點(diǎn)進(jìn)行處理。新增了 BuildBloomFilter 和 InBloomFilter 兩個(gè) UDF。BuildBloomFilter 在 sparkPlan prepare 階段提交任務(wù)構(gòu)造 BloomFilter 并 broadcast 出去，具體的 evaluate 邏輯還是交給 InBloomFilter。另外在 AQE 的reOptimize 階段也新增了規(guī)則 OptimizeBloomFilterJoin，這個(gè)規(guī)則主要是用來(lái)根據(jù)執(zhí)行過(guò)程的 metric 信息更新BuildBloomFilter的expectedNumItems。

可以看到在開(kāi)啟了runtime filter后數(shù)據(jù)量在join前從120億條降至3W條，收益還是相當(dāng)明顯的。

3.2.4 Data skipping

目前B站離線表存儲(chǔ)主要使用 orc、parquet 格式，列式存儲(chǔ)都支持一定程度的 data skipping，比如 orc 有三個(gè)級(jí)別的統(tǒng)計(jì)信息，file/stripe/row group，統(tǒng)計(jì)信息中會(huì)包含count，對(duì)于原始類型的列，還會(huì)記錄 min/max 值，對(duì)于數(shù)值類型的列，也會(huì)記錄 sum 值。在查詢時(shí)，就可以根據(jù)不同粒度的統(tǒng)計(jì)信息以及 index 決定該 file/stripe/row 是否符合條件，不符合條件的直接跳過(guò)。對(duì)于統(tǒng)計(jì)信息及索引的細(xì)節(jié)見(jiàn)orc format  (https://orc.apache.org/specification/ORCv1/)  和 orc index (https://orc.apache.org/docs/indexes.html)  。Parquet 與 orc 類似，也有相應(yīng)的設(shè)計(jì)，具體見(jiàn)parquet format (https://github.com/apache/parquet-format)  和 parquet pageIndex (https://github.com/apache/parquet-format/blob/master/PageIndex.md)  。雖然 orc/parquet 都有 data skipping 的能力，但這種能力非常依賴數(shù)據(jù)的分布。前面提到統(tǒng)計(jì)信息中會(huì)包含每一列的 min/max 值，理論上如果查詢條件(比如> < =)不在這個(gè)范圍內(nèi)，那么這個(gè)file/stripe/row group 就可以被跳過(guò)。但如果數(shù)據(jù)沒(méi)有按照 filter 列排序，那最壞的情況下，可能每個(gè) file/stripe/row group的min/max 值都一樣，這樣就造成任何粒度的數(shù)據(jù)都不能被跳過(guò)。為了增加列式存儲(chǔ) data skipping 效果，可以通過(guò)對(duì)數(shù)據(jù)增加額外的組織，如下：

select
    count(1)
  from
    tpcds.archive_spl_cluster
  where
    log_date = '20211124'
    and state = -16

表 archive_spl，不調(diào)整任何分布與排序

表 archive_spl_order，order by state,avid

通過(guò)對(duì) state 進(jìn)行 order 后 scan 階段數(shù)據(jù)量直接從億級(jí)別降至數(shù)十萬(wàn)級(jí)別。在生產(chǎn)中我們通過(guò)對(duì) SQL 進(jìn)行血緣分析找到那些熱點(diǎn)表及高頻 filter 列，將這些熱列作為 table properties 存入 hms 中，在 Spark 執(zhí)行時(shí)根據(jù)從 hms 中獲取的列信息，通過(guò)相應(yīng)的優(yōu)化規(guī)則，物理計(jì)劃自動(dòng)增加 sort 算子，完成對(duì)數(shù)據(jù)組織。這個(gè)方案是基于列存優(yōu)化數(shù)據(jù)組織來(lái)進(jìn)行 data skipping，目前我們也在往索引方向上進(jìn)一步探索。

3.3 功能性改進(jìn)

3.3.1 對(duì)于ZSTD的支持

Spark 社區(qū)在3.2版本全面支持了 ZSTD 壓縮，為了更好的使用 ZSTD，我們?cè)?Spark3.1  的基礎(chǔ)上引入了社區(qū)的相關(guān) patch。其中也遇到了一些問(wèn)題。在測(cè)試 ZSTD 的過(guò)程中偶然發(fā)現(xiàn)下推到 ORC 的過(guò)濾條件沒(méi)有生效，經(jīng)調(diào)查發(fā)現(xiàn)是 ORC 代碼的 bug，在和社區(qū)討論之后，我們修復(fù)了該 bug并將 patch提交給了社區(qū)：https://issues.apache.org/jira/browse/ORC-1121 。

離線平臺(tái)的 Presto 也承接了很多 ETL 任務(wù)，由于 Presto 使用的是自己實(shí)現(xiàn)的 ORC reader/writer，所以在 Spark 升級(jí) ORC 版本之后，對(duì)一些 Presto 寫出的表，出現(xiàn)了查詢結(jié)果錯(cuò)誤的問(wèn)題。正常情況下，Apache ORC writer 在寫文件時(shí)會(huì)記錄每個(gè) stripe/rowGroup 中每列的統(tǒng)計(jì)信息，如 min/max 等。Apache ORC reader 在讀取文件時(shí)會(huì)根據(jù)這些統(tǒng)計(jì)信息結(jié)合下推的過(guò)濾條件進(jìn)行 stripe/rowGroup 級(jí)別的過(guò)濾。但 Presto ORC writer 在寫文件時(shí)，如果 String 類型的列長(zhǎng)度超過(guò)64 bytes，則這一列不會(huì)記錄 min/max 信息。雖然 Presto ORC reader 可以正常處理這類文件，但 Spark/Hive 使用的 Apache ORC reader 因?yàn)闊o(wú)法正常的反序列化 columnStatistics 得到正確的統(tǒng)計(jì)信息，導(dǎo)致做 stripe/rowGroup 級(jí)別的過(guò)濾時(shí)出現(xiàn)了錯(cuò)誤的結(jié)果。我們也發(fā)現(xiàn)這個(gè)問(wèn)題是由于 ORC 1.6 版本的一次代碼重構(gòu)導(dǎo)致，1.5及之前是沒(méi)有該問(wèn)題的。我們已在內(nèi)部分支修復(fù)了該問(wèn)題，也已將問(wèn)題反饋給社區(qū)。

3.3.2 多格式混合讀兼容

歷史上很多表使用了 text 存儲(chǔ)，在資源上造成了很大的浪費(fèi)，通過(guò)修改表的元信息可以保障新增分區(qū)切換到列存，這就造成了一個(gè)離線表可能存在多種 fileformat 的情況，為了兼容我們修改了 DataSourceScanExec 相關(guān)的邏輯，將reader 的實(shí)例化從基于table元信息粒度細(xì)化到分區(qū)元信息粒度。

3.3.3 轉(zhuǎn)表&小文件合并語(yǔ)法

為了方便用戶修改表的存儲(chǔ)格式和文件壓縮格式我們?cè)谝鎸犹峁┝讼嚓P(guān)語(yǔ)法及具體實(shí)現(xiàn)。用戶可以通過(guò)指定分區(qū)條件對(duì)特定分區(qū)進(jìn)行轉(zhuǎn)換。

CONVERT TABLE target=tableIdentifier
       (convertFormat | compressType)  partitionClause?               #convertTable
MERGE TABLE target=tableIdentifier
       partitionClause?                                               #mergeTable

3.3.4 字段血緣

作業(yè)間的依賴關(guān)系分析、數(shù)據(jù)地圖等業(yè)務(wù)都需要SQL血緣的支持，團(tuán)隊(duì)后續(xù)工作（z-order , analyze , index）也需要依賴血緣，我們通過(guò)注冊(cè)一個(gè) LineageQueryListener 繼承 QueryExecutionListener，在 onSuccess 方法拿到當(dāng)前執(zhí)行的QueryExecution，通過(guò) analyzedLogicalPlan，利用 NamedExpression 的 exprId 映射關(guān)系，對(duì)其進(jìn)行遍歷和解析，構(gòu)建出字段級(jí)血緣（PROJECTION/PREDICATE）和 levelRelation（層級(jí)關(guān)系）。

3.4 基于歷史執(zhí)行的自動(dòng)參數(shù)優(yōu)化（HBO）

Spark 提供了大量的參數(shù)設(shè)置，對(duì)于用戶而言了解這些參數(shù)并使用好需要花費(fèi)很大的代價(jià)，在很多情況下不同的參數(shù)調(diào)優(yōu)對(duì)于 spark 的作業(yè)執(zhí)行和資源消耗會(huì)有很大差異。為了盡可能的適配任務(wù)執(zhí)行，我們預(yù)設(shè)了一組參數(shù)，這種統(tǒng)一配置存在很多問(wèn)題，以內(nèi)存而言為了適配盡可能多的任務(wù)，該值設(shè)置偏大，通過(guò)對(duì)執(zhí)行的分析發(fā)現(xiàn)大量的任務(wù)存在資源浪費(fèi)的問(wèn)題，整體的內(nèi)存利用率僅20%左右。要求每個(gè)用戶成為專家對(duì)作業(yè)進(jìn)行細(xì)致的調(diào)優(yōu)顯然不可能，因此我們?cè)O(shè)計(jì)了 HBO 系統(tǒng)，具體的思路如下圖：

首先對(duì)任務(wù)執(zhí)行的 SQL 進(jìn)行了指紋計(jì)算，通過(guò)指紋來(lái)標(biāo)識(shí)該任務(wù)每天執(zhí)行情況，將每次執(zhí)行中采集到的 metrics 收集后用策略進(jìn)行分析給出相應(yīng)的參數(shù)優(yōu)化建議，在下次執(zhí)行的時(shí)候根據(jù)指紋來(lái)獲取推薦的執(zhí)行參數(shù)，對(duì)于使用默認(rèn)參數(shù)的任務(wù)則進(jìn)行覆蓋，對(duì)于那些用戶指定的參數(shù)則優(yōu)先使用用戶參數(shù)。

• 內(nèi)存優(yōu)化策略：通過(guò)收集每個(gè) executor 的峰值內(nèi)存，如果峰值內(nèi)存占配置內(nèi)存比值低于30%，就推薦使用更少的內(nèi)存來(lái)執(zhí)行此次的計(jì)算，對(duì)于峰值內(nèi)存占比過(guò)高的任務(wù)，則調(diào)大內(nèi)存配置。通過(guò)這個(gè)策略生產(chǎn)上的內(nèi)存使用率提升至50%左右。
• 并行度優(yōu)化策略：生產(chǎn)上開(kāi)啟了動(dòng)態(tài)資源配置，在對(duì)數(shù)據(jù)分析時(shí)發(fā)現(xiàn)有些節(jié)點(diǎn)從分配后就沒(méi)有task執(zhí)行過(guò)，完全浪費(fèi)了節(jié)點(diǎn)的資源，對(duì)于這些任務(wù)會(huì)在下次執(zhí)行的時(shí)候降低 spark.dynamicAllocation.executorAllocationRatio 值來(lái)降低執(zhí)行并行度，此外默認(rèn)提供的 spark.sql.shuffle.partitions 值對(duì)于大任務(wù)來(lái)說(shuō)執(zhí)行并行度不夠，后續(xù)也將進(jìn)行自動(dòng)的調(diào)整。
  
• 優(yōu)化shuffle策略：如上文所講 RSS 對(duì)小任務(wù)存在性能下降的問(wèn)題，通過(guò)對(duì) block size、shuffle 數(shù)據(jù)量的分析，HBO 系統(tǒng)只會(huì)對(duì)那些 shuffle heavy 任務(wù)開(kāi)啟使用 RSS 服務(wù)。
  
• 小文件合并策略：小文件合并會(huì)消耗額外的資源，對(duì)于不存在小文件情況的作業(yè) HBO 系統(tǒng)會(huì)關(guān)閉小文件合并相關(guān)的配置。
  
• 此外平時(shí)工作中一些 feature 的上線也會(huì)依賴該系統(tǒng)進(jìn)行一個(gè)灰度過(guò)程。
  

3.5 Smart Data Manager (SDM)  

Smart Data Manager（SDM）是我們自研的一個(gè)對(duì)數(shù)據(jù)進(jìn)行組織和分析的服務(wù)，通過(guò)對(duì)數(shù)據(jù)的額外處理將我們對(duì) Spark 的一些技改真正落地。它的整體架構(gòu)如圖，目前提供了如下的幾個(gè)數(shù)據(jù)組織和分析能力：

• 表存儲(chǔ)和壓縮方式的轉(zhuǎn)換：將表從 Text 存儲(chǔ)轉(zhuǎn)換為 ORC 或 Parquet 存儲(chǔ)，將壓縮類型從 None 或 Snappy 轉(zhuǎn)換為 ZSTD 可以帶來(lái)不錯(cuò)的存儲(chǔ)和性能收益，SDM 提供了按分區(qū)對(duì)表異步進(jìn)行轉(zhuǎn)換的能力。
• 數(shù)據(jù)重組織：在分區(qū)內(nèi)部按列對(duì)數(shù)據(jù)進(jìn)行 order/zorder 組織可以有效的提高 data skipping 的效果，新增分區(qū)通過(guò)查詢 table properties 中的排序列 meta 來(lái)改寫執(zhí)行計(jì)劃應(yīng)用，存量分區(qū)就可以通過(guò) SDM 重刷。
  
• Statistics 的統(tǒng)計(jì)：開(kāi)啟 CBO 時(shí)需要依賴對(duì)表統(tǒng)計(jì)信息的收集，在對(duì) hive 表的列進(jìn)行索引時(shí)也依賴收集到的列基數(shù)和操作信息選擇合適的索引類型，通過(guò) sdm 監(jiān)聽(tīng) hms 的 partition 事件就可以在分區(qū)更新時(shí)異步完成信息采樣。
  
• 小文件合并：對(duì)有小文件較多的分區(qū)和表異步進(jìn)行小文件合并，減少 namenode 的壓力
  
• Hive 表索引：通過(guò)分析血緣信息得到熱表熱列上的高頻操作(點(diǎn)查，范圍查詢)，基于此在分區(qū)文件層面異步的建立索引來(lái)加速查詢。
  
• 血緣解析：解析語(yǔ)句，分析字段血緣，吐出 UDF 血緣、算子(order by / sort by / group by...)影響關(guān)系等
  

對(duì)數(shù)據(jù)進(jìn)行重組織時(shí)會(huì)涉及到對(duì)數(shù)據(jù)的讀寫，為了防止對(duì)生產(chǎn)作業(yè)的影響我們?cè)谶M(jìn)行操作時(shí)會(huì)修改相關(guān)表的 Table Properties 增加鎖表標(biāo)記，各個(gè)計(jì)算引擎適配實(shí)現(xiàn)了類 Hive 的鎖管理機(jī)制，由 Hive metastore 統(tǒng)一作為 lock manager，在對(duì)表和分區(qū)并發(fā)操作場(chǎng)景下，做到對(duì)用戶完全透明。

4. Hive Meta Store 上的優(yōu)化

B站使用 HMS(Hive MetaStore)管理所有的離線表元信息，整個(gè)的離線計(jì)算的可用性都依賴 HMS 的穩(wěn)定性。業(yè)務(wù)方在使用分區(qū)表時(shí)存在不少4級(jí)及以上分區(qū)的情況，有多個(gè)表分區(qū)數(shù)超百萬(wàn)。分區(qū)元信息龐大單次分區(qū)獲取代價(jià)高，原生 HMS 基于單個(gè) MySQL 實(shí)例存在性能瓶頸。

4.1 MetaStore Federation

隨著多機(jī)房業(yè)務(wù)的推進(jìn)，獨(dú)立業(yè)務(wù)的 HDFS 數(shù)據(jù)和計(jì)算資源已經(jīng)遷移到新機(jī)房，但是 HIVE 元數(shù)據(jù)仍在原有機(jī)房的 Mysql 中，這時(shí)候如果發(fā)生機(jī)房間的網(wǎng)絡(luò)分區(qū)，就會(huì)影響新機(jī)房的任務(wù)。

為了解決上述問(wèn)題，我們進(jìn)行了方案調(diào)研，有兩種方案供我們選擇：

• WaggleDance
• HMS Federation
  

4.1.1 WaggleDance

WaggleDance是開(kāi)源的一個(gè)項(xiàng)目(https://github.com/ExpediaGroup/waggle-dance)，該項(xiàng)目主要是聯(lián)合多個(gè) HMS 的數(shù)據(jù)查詢服務(wù)，實(shí)現(xiàn)了一個(gè)統(tǒng)一的路由接口解決多套 HMS 環(huán)境間的元數(shù)據(jù)共享問(wèn)題。并且 WaggleDance 支持 HMS Client的接口調(diào)用。主要是通過(guò) DB，把請(qǐng)求路由到對(duì)應(yīng)的 HMS。

4.1.2 HMS Federation

HMS Federation 是解決多機(jī)房場(chǎng)景下的 HIVE 元數(shù)據(jù)存儲(chǔ)問(wèn)題，HIVE 元數(shù)據(jù)和 HDFS 數(shù)據(jù)存儲(chǔ)在同一個(gè)機(jī)房，并且允許跨機(jī)房訪問(wèn) HIVE 元數(shù)據(jù)。比如主站業(yè)務(wù)的 HDFS 數(shù)據(jù)存放在 IDC1，那么主站業(yè)務(wù) HDFS 數(shù)據(jù)對(duì)應(yīng)的 HIVE 元數(shù)據(jù)就存在IDC1 的 Mysql，同樣直播業(yè)務(wù)的 HDFS 數(shù)據(jù)和 HIVE 元數(shù)據(jù)都存放在 IDC2。

同時(shí) HMS Federation 也提供了 Mysql 的橫向擴(kuò)容能力，允許一個(gè)機(jī)房可以有多個(gè) Mysql 來(lái)存放 HIVE 元數(shù)據(jù)，如果單個(gè) Mysql 的壓力過(guò)大，可以把單個(gè) Mysql 的數(shù)據(jù)存放到多個(gè) Mysql 里面，分擔(dān) Mysql 的壓力。比如主站業(yè)務(wù)的 HIVE 庫(kù)，zhu_zhan 和 zhu_zhan_tmp，可以分別放在 idc1-mysql1 和 idc1-mysql2。

我們?cè)?HMS Federation 中加入了一個(gè) StateStore 的角色，該角色可以理解為一個(gè)路由器，HMS 在查詢 Hive 庫(kù)/表/分區(qū)之前，先問(wèn) StateStore 所要訪問(wèn)的 HIVE 元信息存放在哪一個(gè) Mysql 中，獲取到了對(duì)應(yīng)的 Mysql 后，構(gòu)建相應(yīng)的ObjectStore，進(jìn)行 SQL 拼接或者是利用 JDO 查詢后端 Mysql。

4.1.3 HMS Federation 與 WaggleDance 的對(duì)比

數(shù)據(jù)遷移

我們的主要目的是實(shí)現(xiàn) HIVE 元數(shù)據(jù)按業(yè)務(wù)劃分到各自 IDC 的 Mysql

• WaggleDance 并沒(méi)有提供相應(yīng)元數(shù)據(jù)遷移工具，要遷移需要停止整個(gè) HIVE 庫(kù)新建表/分區(qū)，才能夠開(kāi)始遷移過(guò)去，對(duì)業(yè)務(wù)影響較大。
• HMS Federation 可以按表的粒度遷移，對(duì)業(yè)務(wù)影響較小，并且可以指定某個(gè) HIVE 庫(kù)下，新建表在新的 Mysql，舊的等待著鎖表遷移。
  

運(yùn)維復(fù)雜度

• WaggleDance 方案需要不同的 HMS，配置不同的 Mysql 地址，增加了 HMS 配置的復(fù)雜度。WaggleDance 是一個(gè)獨(dú)立的服務(wù)，為了保證可用性，運(yùn)維復(fù)雜度會(huì)再一次提升。
• HMS Fedration 是 HMS 的功能升級(jí)，在 HMS 代碼上開(kāi)發(fā)，并且使用統(tǒng)一的配置。
  

綜合上述對(duì)比，我們最終選擇了 HMS Federation 的方案。通過(guò)修改 HMS 的代碼，實(shí)現(xiàn)元數(shù)據(jù)跨 Mysql 存儲(chǔ)。

4.2 MetaStore 請(qǐng)求追蹤和流量控制

HMS 在處理 getPartitions 相關(guān)請(qǐng)求的時(shí)候，如果拉取的分區(qū)數(shù)量非常多，會(huì)給 HMS 的堆內(nèi)存，以及后端的 Mysql 帶來(lái)很大的壓力，導(dǎo)致 HMS 服務(wù)響應(yīng)延遲。

為了能夠快速的定位到有問(wèn)題的任務(wù)，我們?cè)?Driver 中將 Job 相關(guān)的信息保存到 Hadoop CallerContext 中，在調(diào)用 HMS 接口的時(shí)候?qū)?CallerContext 中的相關(guān)屬性設(shè)置到 EnvironmentContext 中透?jìng)鞯?HMS 端，同時(shí)擴(kuò)展了所有g(shù)etPartitions 相關(guān)的接口支持傳遞 EnvironmentContext，EnvironmentContext 中的 properties 會(huì)在 HMS 的 audit log 中打印出來(lái)，方便問(wèn)題任務(wù)的定位。

同時(shí)為了提高 HMS 服務(wù)的穩(wěn)定性，我們?cè)?HMS 端也做了接口的限流和主動(dòng)關(guān)閉大查詢。對(duì)于限流，我們新增了一個(gè) TrafficControlListener，當(dāng)接口被調(diào)用的時(shí)候會(huì)以 function 和 user 為單位記錄 Counters 保存在該 Listener 中，同時(shí)在該Listener 中啟動(dòng)采集 used memory 和 counters 的線程，當(dāng)平均使用內(nèi)存達(dá)到閾值時(shí)，檢查接口的QPS，如果qps達(dá)到閾值會(huì)讓調(diào)用接口的線程 sleep 一段時(shí)間，下一次檢查通過(guò)或者達(dá)到最大等待時(shí)間后放行。HMS 也有可能因?yàn)?getPartitions 方法返回的分區(qū)數(shù)量太大導(dǎo)致內(nèi)存被打滿，一方面我們限制了 getPartitions 從 mysql 返回的分區(qū)數(shù)量，超過(guò)一定數(shù)量就直接拒絕該請(qǐng)求，另一方面我們?cè)?TProcessor 中以 threadId 和 socket 為 key 和 value 保存當(dāng)前的連接，在檢查 partition 數(shù)量時(shí)我們也按照 threadId 和 num partitions 為 key 和 value 保存 partition 的 cost，當(dāng) HMS 平均使用內(nèi)存達(dá)到閾值超過(guò)一定時(shí)間后，會(huì)選擇 num partitions 最大的 threadId，再根據(jù) threadId 獲取對(duì)應(yīng)的連接，主動(dòng) close 該連接，來(lái)緩解內(nèi)存壓力。

5. 未來(lái)的一些工作

調(diào)研不落地的 Remote Shuffle Service 來(lái)更好的適配 K8S 混部的場(chǎng)景

使用向量化技術(shù)加速 Spark 的執(zhí)行引擎，提升計(jì)算性能

增強(qiáng)自動(dòng)排錯(cuò)診斷系統(tǒng)，提升平臺(tái)用戶體驗(yàn)

我們會(huì)和業(yè)界同行和開(kāi)源社區(qū)保持密切技術(shù)交流，在服務(wù)好內(nèi)部用戶作業(yè)的同時(shí)，也會(huì)積極反饋社區(qū)，共建社區(qū)生態(tài)。