01 背景

隨著B站業(yè)務(wù)的高速發(fā)展，業(yè)務(wù)數(shù)據(jù)的生產(chǎn)速度變得越來越快，離線集群規(guī)?？焖倥蛎?，既有機(jī)房?jī)?nèi)的機(jī)位急劇消耗，在可預(yù)見的不久的將來會(huì)達(dá)到機(jī)房容量上限，阻塞業(yè)務(wù)的發(fā)展。因此，如何解決單機(jī)房容量瓶頸成為了我們亟待解決的問題。

目前，針對(duì)機(jī)房容量問題的解決方案業(yè)界主要有以下兩種：

1） 集群整體搬遷至更高容量的機(jī)房（scale up） 。該方案是一種縱向擴(kuò)容方案，即將現(xiàn)有集群搬遷至容量更大的機(jī)房，從而提供集群擴(kuò)展的空間?，F(xiàn)實(shí)中，集群遷移一般不能影響業(yè)務(wù)的發(fā)展，即保證不停機(jī)，因此，遷移過程中需要兩個(gè)規(guī)模相近的集群做全量遷移，或者需要一個(gè)具有一定規(guī)模的過渡集群，分批次遷移；對(duì)于大規(guī)模（tens of thousands）集群來說，遷移的經(jīng)濟(jì)成本巨大；另外，遷移后的新機(jī)房會(huì)有再次達(dá)到容量上限的風(fēng)險(xiǎn)。

2） 多機(jī)房方案（scale out） ，即一個(gè)機(jī)房容量有限，擴(kuò)展為多個(gè)機(jī)房，同時(shí)對(duì)既有架構(gòu)進(jìn)行一定的改造，保證用戶視角仍像是一個(gè)機(jī)房。此舉可依據(jù)業(yè)務(wù)需要，采用靈活的方式增量擴(kuò)容，從而一定程度上避免容量冗余問題。然而，該方案會(huì)存在跨機(jī)房數(shù)據(jù)交互，而機(jī)房間網(wǎng)絡(luò)帶寬一般也存在瓶頸；同時(shí)，網(wǎng)絡(luò)的抖動(dòng)或斷網(wǎng)可能造成跨機(jī)房業(yè)務(wù)出現(xiàn)異常。因此，該方案需要考慮/解決網(wǎng)絡(luò)帶寬不足及網(wǎng)絡(luò)抖動(dòng)/斷網(wǎng)問題帶來的影響，技術(shù)成本較集群整體搬遷方案要高。

就我們目前自建機(jī)房的情況來看，中短期暫無清退既有機(jī)房（全部搬遷至新機(jī)房）的計(jì)劃，從長(zhǎng)期來看也會(huì)存在多個(gè)機(jī)房；另外，比起方案2的技術(shù)成本，我們更難接受方案1的經(jīng)濟(jì)成本和容量風(fēng)險(xiǎn)。因此，方案2是我們解決機(jī)房容量問題首選方案。

02 多機(jī)房方案

2.1 面臨的問題

上文提到多機(jī)房方案面臨帶寬等網(wǎng)絡(luò)問題，多機(jī)房方案的設(shè)計(jì)受其制約。

帶寬瓶頸

離線場(chǎng)景主要是批處理場(chǎng)景，是對(duì)海量歷史數(shù)據(jù)進(jìn)行離線分析/處理的場(chǎng)景，該場(chǎng)景對(duì)延遲不敏感，但由于其處理數(shù)據(jù)量巨大對(duì)網(wǎng)絡(luò)帶寬等資源消耗較大；另外，生產(chǎn)場(chǎng)景中作業(yè)數(shù)量一般較多且執(zhí)行時(shí)間不受控，若兩個(gè)機(jī)房的主機(jī)只是簡(jiǎn)單疊加在一起做為一個(gè)集群來用，可能會(huì)存在大量的跨機(jī)房互訪，產(chǎn)生大量的隨機(jī)流量打滿有限的跨機(jī)房帶寬, 此時(shí)除離線自身受影響外, 還可能對(duì)其它跨機(jī)房業(yè)務(wù)造成影響 。因此，如何防止跨機(jī)房隨機(jī)流量打滿跨機(jī)房帶寬是多機(jī)房方案要解決的一個(gè)重要問題。

網(wǎng)絡(luò)抖動(dòng)&連通性

跨城網(wǎng)絡(luò)會(huì)受供應(yīng)商服務(wù)質(zhì)量影響（或施工影響）造成抖動(dòng)（或斷網(wǎng)）, 與機(jī)房?jī)?nèi)CLOS架構(gòu)的網(wǎng)絡(luò)質(zhì)量相比會(huì)低很多。若兩個(gè)機(jī)房的主機(jī)當(dāng)做為一個(gè)集群來用，如圖1 HDFS示例，當(dāng)網(wǎng)絡(luò)抖動(dòng)時(shí)，不但會(huì)導(dǎo)致跨機(jī)房讀寫延遲增加，還會(huì)影響DN的IBR等過程，造成服務(wù)性能和穩(wěn)定性下降；當(dāng)網(wǎng)絡(luò)出現(xiàn)嚴(yán)重問題造成斷網(wǎng)時(shí)，會(huì)導(dǎo)致異地機(jī)房數(shù)據(jù)不可用，還會(huì)導(dǎo)致異地機(jī)房DN失聯(lián)，造成大量Block低于預(yù)期副本數(shù)，觸發(fā)NN大量補(bǔ)副本等問題。因此，如何降低網(wǎng)絡(luò)抖動(dòng)及網(wǎng)絡(luò)連通性問題帶來的影響是多機(jī)房方案要解決的另外一個(gè)不可忽視的問題。

圖1 HDFS 架構(gòu)

2.2 設(shè)計(jì)選型

如上所述，多機(jī)房的主要矛盾是跨機(jī)房網(wǎng)絡(luò)帶寬不足、穩(wěn)定性差與離線海量數(shù)據(jù)處理任務(wù)高效產(chǎn)出之間的矛盾，解決該主要矛盾面臨的核心問題是如何減少跨機(jī)房帶寬的消耗，以及如何降低網(wǎng)絡(luò)穩(wěn)定性問題帶來的影響。

經(jīng)調(diào)研，單元化架構(gòu)是為解決多地多中心問題演進(jìn)而來的部署架構(gòu)，其中，單元是指一個(gè)能完成所有業(yè)務(wù)操作的自包含集合，在這個(gè)集合中包含了業(yè)務(wù)所需的所有服務(wù)，以及分配給這個(gè)單元的數(shù)據(jù) [1-2] 。按照單元化的思路，在多機(jī)房場(chǎng)景中，每個(gè)機(jī)房可以作為一個(gè)單元，每個(gè)單元內(nèi)提供作業(yè)執(zhí)行所需要的全部服務(wù)以及數(shù)據(jù)，保證作業(yè)在單元內(nèi)完成，從而解決上述多機(jī)房面臨的核心問題；單元化拆分后任何一個(gè)單元的故障只會(huì)影響局部，不會(huì)造成整體癱瘓；在選定采用單元化思想來設(shè)計(jì)了多機(jī)房方案之后， 多機(jī)房方案的核心問題就限定在了如何決定作業(yè)與數(shù)據(jù)放置，以及如何讓作業(yè)訪問距離近的數(shù)據(jù)，來降低跨機(jī)房帶寬的消耗及網(wǎng)絡(luò)穩(wěn)定性問題帶來的影響。

帶著上面的核心問題，我們調(diào)研了業(yè)界大廠的多機(jī)房解決方案 [3-7] 。這些方案在計(jì)算層面為防止Shuffle等中間結(jié)果數(shù)據(jù)造成跨機(jī)房流量，每個(gè)機(jī)房均獨(dú)立部署了計(jì)算集群，在該層面均符合單元化思想；但在存儲(chǔ)存面存在分歧，如圖2所示，依據(jù)數(shù)據(jù)和異地機(jī)房的數(shù)據(jù)副本是否屬于同一組NameSpace (NS)，大體可以分為多機(jī)房單集群方案和多機(jī)房多集群方案。

圖2 多機(jī)房方案分類

[3-5] 采用了多機(jī)房單集群方案，該方案中采用Block級(jí)的數(shù)據(jù)副本，數(shù)據(jù)和數(shù)據(jù)副本同屬于一組NS，無數(shù)據(jù)一致性問題，但因NS只能在其中一個(gè)機(jī)房，無法有效應(yīng)對(duì)網(wǎng)絡(luò)連通性問題，且Namenode異地副本管理（BlockPlacementPolicy）和相關(guān)工具（Mover, Balancer等）改造成本較大，另外該方案可擴(kuò)展性也受單集群規(guī)模制約。

[6-7] 采用了多機(jī)房多集群方案，整體符合單元化思想。其中 [6] 應(yīng)用于云梯遷機(jī)房場(chǎng)景，它首先在同機(jī)房中通過Fast Copy將文件元數(shù)據(jù)分離到兩個(gè)NS，然后再通過同NS內(nèi)DN到DN的跨機(jī)房Copy將數(shù)據(jù)復(fù)制到遠(yuǎn)程機(jī)房，該方案在一定程度上可以有效應(yīng)對(duì)跨機(jī)房網(wǎng)絡(luò)風(fēng)險(xiǎn)，但因存在兩次copy時(shí)效性上難以保障，另外也存在異地的數(shù)據(jù)節(jié)點(diǎn)，因此本質(zhì)上也存在多機(jī)房單集群方案改造成本和擴(kuò)展性問題；[7] 阿里Yugong（Yugong: Geo-Distributed Data and Job Placement at Scale）基于MetaStore針對(duì)分區(qū)表場(chǎng)景，通過調(diào)整作業(yè)放置和數(shù)據(jù)放置來降低跨機(jī)房帶寬的消耗；如圖3所示，計(jì)算A、B存在跨機(jī)房訪問行為，通過調(diào)整（互換）計(jì)算A、B的放置位置可以有效減少跨機(jī)房訪問流量；計(jì)算C、D同時(shí)跨機(jī)房消費(fèi)同一份數(shù)據(jù)3, 若通過數(shù)據(jù)復(fù)制的方式將數(shù)據(jù)3復(fù)制到機(jī)房2， 讓C、D依賴數(shù)據(jù)3在機(jī)房2中的副本，則可以減少一次跨機(jī)房消費(fèi)數(shù)據(jù)流量。但對(duì)于我們采用開源大數(shù)據(jù)架構(gòu)的場(chǎng)景來說，需要改造（分屬于多個(gè)子部門的）多種計(jì)算框架來適配其基于MetaStore的數(shù)據(jù)副本管理和數(shù)據(jù)路由，改造實(shí)施成本較大；另外，其基于MetaStore的設(shè)計(jì)只能解決表（SQL）場(chǎng)景的多機(jī)房問題，也不能覆蓋我們對(duì)非表場(chǎng)景提供多機(jī)房支持的需求；不過，該方案中通過“作業(yè)放置-數(shù)據(jù)復(fù)制”來解決帶寬瓶頸問題的思路非常值得我們借鑒。

圖3 任務(wù)跨機(jī)房隨機(jī)分布

綜上，我們參考Yugong“作業(yè)放置-數(shù)據(jù)復(fù)制”的思路，采用有限的單元化思想設(shè)計(jì)多機(jī)房方案；如圖4所示，每個(gè)機(jī)房部署一套獨(dú)立的完整的集群（YARN&HDFS），為作業(yè)在一個(gè)機(jī)房?jī)?nèi)執(zhí)行提供最基本的服務(wù)保障，從而在跨機(jī)房網(wǎng)絡(luò)出現(xiàn)異常時(shí)，降低影響范圍；同時(shí)，通過合理的作業(yè)放置和有計(jì)劃的數(shù)據(jù)復(fù)制，消除跨機(jī)房隨機(jī)訪問流量及跨機(jī)房數(shù)據(jù)重復(fù)消費(fèi)等問題，來達(dá)到降低帶寬消耗的目的；另外，我們結(jié)合內(nèi)部的基礎(chǔ)設(shè)施情況，以及滿足表和非表兩種場(chǎng)景的需求，我們選擇了基于擴(kuò)展HDFS Router（RBF）多掛載點(diǎn)來實(shí)現(xiàn)數(shù)據(jù)副本管理和數(shù)據(jù)路由功能，并通過Client IP感知自動(dòng)將數(shù)據(jù)請(qǐng)求路由至較近的機(jī)房；還有為解決數(shù)據(jù)復(fù)制帶來的一致性問題引入了Version服務(wù)等，圖中涉及組件將在實(shí)現(xiàn)部分進(jìn)行介紹。

圖4 多機(jī)房架構(gòu)

2.3 總體流程

圖5展示了以Hive作業(yè)為例的在上述設(shè)計(jì)思路下的總體流程，圖中綠色模塊為我們新增或改造組件。首先，通過周期性的分析作業(yè)間依賴關(guān)系及依賴的數(shù)據(jù)大小，確定作業(yè)放置位置信息并進(jìn)行持久化（DataManager用于管理作業(yè)放置信息等），B站的作業(yè)調(diào)度平臺(tái)（Archer和Airflow）提交作業(yè)時(shí)，先獲取作業(yè)的放置機(jī)房信息，并檢查預(yù)期放置機(jī)房的數(shù)據(jù)副本是否Ready，若Ready則提交作業(yè)，否則，阻塞提交，等待數(shù)據(jù)復(fù)制服務(wù)完成復(fù)制數(shù)據(jù)；其次，作業(yè)調(diào)度提交后，拉起Hive/Spark Driver生成可執(zhí)行計(jì)劃，向預(yù)期DC的Yarn集群提交Job，等待拉起Job，同時(shí)我們?cè)赮arn層面也做了改造，基于Yarn Federation架構(gòu)，實(shí)現(xiàn)了基于app tag和隊(duì)列的機(jī)房調(diào)度策略，這個(gè)在下文也會(huì)介紹; 最后，被拉起的作業(yè)請(qǐng)求HDFS數(shù)據(jù)，HDFS Router依據(jù)Client IP所屬的DC信息，自動(dòng)將請(qǐng)求路由到距離Client較近的數(shù)據(jù)復(fù)本所在機(jī)房的NS, 并將結(jié)果返回Client。

圖5 多機(jī)房作業(yè)調(diào)度執(zhí)行流程

多機(jī)房核心流程包括作業(yè)放置、數(shù)據(jù)復(fù)制、數(shù)據(jù)路由、版本控制、數(shù)據(jù)限流、跨機(jī)房流量分析等幾個(gè)階段，上述Job提交流程并未完全涵蓋，下文實(shí)現(xiàn)部分我們將對(duì)所有階段進(jìn)行詳細(xì)說明。

03 多機(jī)房方案實(shí)現(xiàn)

下面章節(jié)會(huì)對(duì)多機(jī)房核心環(huán)節(jié)進(jìn)行介紹， 包括作業(yè)放置、數(shù)據(jù)復(fù)制、數(shù)據(jù)路由，以及為保障數(shù)據(jù)副本一致性引入的數(shù)據(jù)版本服務(wù)和帶寬控制的限流服務(wù)，并引入事后的跨機(jī)房流量分析工具，用以發(fā)現(xiàn)預(yù)期外的跨機(jī)房行為指導(dǎo)調(diào)整。

3.1 作業(yè)放置

a. 依賴分析

大數(shù)據(jù)離線場(chǎng)景，作業(yè)數(shù)量多，作業(yè)之間依賴復(fù)雜。比如，大數(shù)據(jù)離線報(bào)表處理業(yè)務(wù)，從數(shù)據(jù)采集，清洗，到各個(gè)層級(jí)的報(bào)表的匯總運(yùn)算，到最后數(shù)據(jù)導(dǎo)出到外部業(yè)務(wù)系統(tǒng)，一個(gè)完整的業(yè)務(wù)流程，可能涉及到成百上千個(gè)相互交叉依賴關(guān)聯(lián)的作業(yè)。就作業(yè)放置來說，對(duì)復(fù)雜作業(yè)依賴的管理和分析工作至關(guān)重要， 而如我們自研的調(diào)度平臺(tái)Archer等DAG工作流類調(diào)度系統(tǒng)，自身具有較強(qiáng)的作業(yè)依賴管理能力，因此，我們僅需要聚焦作業(yè)依賴分析以確定要遷移的業(yè)務(wù)。

我們依據(jù)作業(yè)間依賴關(guān)系及需要處理的數(shù)據(jù)大小，基于社區(qū)發(fā)現(xiàn)（Community Detection）探索了一種考慮跨機(jī)房帶寬代價(jià)的作業(yè)關(guān)系鏈劃分模型。該模型首先依據(jù)調(diào)度系統(tǒng)管理的作業(yè)間的依賴關(guān)系構(gòu)建DAG圖， 然后從DAG圖中圈出相對(duì)高內(nèi)聚（相對(duì)比較閉環(huán)）的業(yè)務(wù)子單元，最后結(jié)合相互依賴的子單元間的數(shù)據(jù)量選擇出的可以遷移的子單元；如圖6所示的簡(jiǎn)單DAG, 我們假定圖中正方形代表計(jì)算，圓形代表數(shù)據(jù)，圓的大小代表數(shù)據(jù)大小，則我們以虛線作為劃分邊界將DAG分成兩個(gè)子單元，分別調(diào)度到兩個(gè)機(jī)房，則可滿足數(shù)據(jù)傳輸代價(jià)小的目標(biāo)。當(dāng)然，整個(gè)過程除了考慮跨機(jī)房數(shù)據(jù)訪問代價(jià)外，還需要考慮機(jī)房計(jì)算和存儲(chǔ)資源是否可以滿足需求。

圖6 依賴關(guān)系劃分

一般而言，實(shí)際生產(chǎn)中的ETL等周期性調(diào)度作業(yè)相對(duì)比較穩(wěn)定， 不會(huì)頻繁發(fā)生變化，甚至部分作業(yè)不會(huì)出現(xiàn)變化，因此，確定Job放置在那個(gè)機(jī)房的的依賴分析過程可以以天或周為單位周期性的離線計(jì)算產(chǎn)生；另外，從管理的角度來看，公司一般會(huì)有多個(gè)相對(duì)比較獨(dú)立的業(yè)務(wù)部門，每個(gè)業(yè)務(wù)部門又會(huì)垂直的劃分出多個(gè)業(yè)務(wù)子單元，業(yè)務(wù)內(nèi)的作業(yè)間聯(lián)系緊密程度遠(yuǎn)大于業(yè)務(wù)之間；同時(shí)，業(yè)務(wù)（單元）也是資源管理單元，以及多機(jī)房落地實(shí)施過程中的溝通單元；因此，在實(shí)踐中往往是以業(yè)務(wù)單元為邊界進(jìn)行依賴劃分。

b. 作業(yè)放置

我們的生產(chǎn)環(huán)境中存在多個(gè)作業(yè)調(diào)度平臺(tái)，如Archer、Airflow等平臺(tái)，將Job放置在那個(gè)機(jī)房的信息維護(hù)在任一平臺(tái)都不能涵蓋所有作業(yè)， 因此我們引入DataManager服務(wù)（在整個(gè)體系中的位置見圖4）作為接入層，用來管理作業(yè)放置的IDC信息和需要進(jìn)行數(shù)據(jù)復(fù)制的路徑信息，Archer/Airflow等調(diào)度平臺(tái)通過對(duì)接該服務(wù)來接入多機(jī)房體系；下面以自研DAG調(diào)度平臺(tái)Archer為例描述工作流程如下：

前置工作：Archer 通過DataManager接口設(shè)置作業(yè)的放置位置信息，以及依賴數(shù)據(jù)的路徑pattern、范圍、生命周期等信息。

Archer訪問DataManager確定作業(yè)放置的IDC信息，并為作業(yè)選擇符合IDC作業(yè)配置信息。

Archer詢問Job在該IDC的數(shù)據(jù)是否Ready，若Ready，則設(shè)置app tag為該IDC并通過Yarn RMProxy向計(jì)算集群提供作業(yè)；否則，掛起并等待數(shù)據(jù)Ready后嘗試重新提交；其中數(shù)據(jù)是否Ready，是通過DataManager轉(zhuǎn)發(fā)請(qǐng)求至數(shù)據(jù)復(fù)制服務(wù)得到。

另外由于我們的業(yè)務(wù)部門和Yarn上的一級(jí)隊(duì)列做了一一映射，所以一旦某個(gè)業(yè)務(wù)部門的數(shù)據(jù)整體遷移到新機(jī)房后，我們會(huì)在RMProxy中設(shè)置該部門對(duì)應(yīng)的queue mapping策略到新機(jī)房，這樣無論是從調(diào)度平臺(tái)還是其他用戶客戶端提交的Job即使沒有接入DataManager也能正確路由到新機(jī)房的計(jì)算集群，同時(shí)回收老機(jī)房的計(jì)算和存儲(chǔ)資源。

圖7 Yarn調(diào)度策略

3.2 數(shù)據(jù)復(fù)制

a. 復(fù)制服務(wù)

作業(yè)放置會(huì)將有聯(lián)系緊密的Job放在一個(gè)機(jī)房，以減少跨機(jī)房訪問，進(jìn)而減少跨機(jī)房網(wǎng)絡(luò)帶寬消耗；對(duì)于無法消除的跨機(jī)房依賴，特別是異地機(jī)房使用頻次大于1的數(shù)據(jù)，需要異地機(jī)房也存在數(shù)據(jù)副本，以降低網(wǎng)絡(luò)帶寬消耗；因此，我們提供了數(shù)據(jù)復(fù)制服務(wù)來進(jìn)行副本復(fù)制。

數(shù)據(jù)復(fù)制服務(wù)基于社區(qū)提供的DistCp工具實(shí)現(xiàn)， 并在正確性、原子性、冪等性、傳輸效率等方面作了增強(qiáng)， 同時(shí)支持流控、多租戶傳輸優(yōu)先級(jí)（高優(yōu)作業(yè)能得到更多跨機(jī)房流量和計(jì)算資源配額），副本生命周期管理等功能。

b. 復(fù)制流程

數(shù)據(jù)復(fù)制主要針對(duì)有規(guī)律的周期性調(diào)度作業(yè)進(jìn)行，這類作業(yè)一般比較固定，通過對(duì)作業(yè)歷史運(yùn)行記錄進(jìn)行分析即可推測(cè)出作業(yè)的輸入輸出情況，包括數(shù)據(jù)路徑和使用的數(shù)據(jù)范圍（防止長(zhǎng)時(shí)間跨度回刷任務(wù)大量復(fù)制）等信息。因此，當(dāng)確定好待遷移的作業(yè)后，可以提煉出數(shù)據(jù)路徑規(guī)則（rules），并持久化到DataManager的規(guī)則庫(kù)中（規(guī)則庫(kù)會(huì)隨作業(yè)放置的變化而進(jìn)行周期性更新）。

然后，針對(duì)不同的場(chǎng)景使用規(guī)則庫(kù)進(jìn)行路徑抽取，下面以Hive表場(chǎng)景為例描述數(shù)據(jù)復(fù)制流程，如圖8所示， 首先收集Hive MetaStore的掛載表/分區(qū)相關(guān)的Event信息至Kafka服務(wù)，然后通過實(shí)時(shí)Flink任務(wù)清洗出符合上述規(guī)則庫(kù)中規(guī)則的路徑，當(dāng)檢測(cè)到熱點(diǎn)表的新分區(qū)生成后，交由數(shù)據(jù)復(fù)制服務(wù)（DRS）進(jìn)行傳輸，生成異地機(jī)房副本，DRS本質(zhì)上是一個(gè)DistCp作業(yè)的管理服務(wù)，在傳輸完成后由數(shù)據(jù)復(fù)制服務(wù)持久化副本信息（包括路徑、版本、TTL等），以對(duì)副本數(shù)據(jù)進(jìn)行全生命周期管理（刪除過期的跨機(jī)房副本，釋放存儲(chǔ)空間），目前B站線上有100+張Hive熱點(diǎn)表路徑設(shè)置了跨機(jī)房副本策略。

圖8 數(shù)據(jù)復(fù)制流程

上述復(fù)制流程采用自動(dòng)發(fā)現(xiàn)主動(dòng)復(fù)制的策略，可以快速捕獲并準(zhǔn)備數(shù)據(jù)副本，經(jīng)過統(tǒng)計(jì)在我們的生產(chǎn)中數(shù)據(jù)副本延遲的PT90可以控制在1min以內(nèi), PT99 在5min以內(nèi)，可以有效滿足離線場(chǎng)景的業(yè)務(wù)需要；然而，上述自動(dòng)發(fā)現(xiàn)主動(dòng)復(fù)制的策略，可以有效解決增量數(shù)據(jù)副本的問題，但對(duì)于待遷移作業(yè)來說，可能還依賴較長(zhǎng)一段時(shí)間的存量數(shù)據(jù)，針對(duì)該問題，我們除了采用提前啟動(dòng)復(fù)制流程的方式準(zhǔn)備存量數(shù)據(jù)外，還針對(duì)需要快速遷移的場(chǎng)景引入了基于Snapshot的數(shù)據(jù)遷移策略進(jìn)行初始復(fù)制，因Snapshot為社區(qū)成熟技術(shù)不再綴述。

3.3 數(shù)據(jù)路由

上小節(jié)介紹的數(shù)據(jù)拷貝后雙機(jī)房均會(huì)存在某路徑的數(shù)據(jù)副本，當(dāng)作業(yè)放置到IDC后如何定位到正確的數(shù)據(jù)是數(shù)據(jù)路由服務(wù)要解決的關(guān)鍵問題；

我們?cè)?nbsp;《HDFS在B站的探索和實(shí)踐》 中提到的基于HDFS Router的多掛載點(diǎn)實(shí)現(xiàn)的MergeFs的基礎(chǔ)上，實(shí)現(xiàn)了鏡像掛載點(diǎn)來實(shí)現(xiàn)數(shù)據(jù)路由功能。為方便描述，我們約定原始數(shù)據(jù)為主數(shù)據(jù)， 傳輸?shù)疆惖貦C(jī)房的數(shù)據(jù)為副本數(shù)據(jù)（也稱為鏡像數(shù)據(jù)，該數(shù)據(jù)只允許讀取和刪除），并且約定鏡像掛載點(diǎn)中第一掛載點(diǎn)為主數(shù)據(jù)，之后的掛載點(diǎn)為副本數(shù)據(jù)（理論上可以擴(kuò)展多個(gè)機(jī)房），為了在路由層面做到對(duì)用戶透明，我們?cè)阽R像掛載點(diǎn)的處理邏輯中，增加了請(qǐng)求來源的IP位置感知功能，該功能能過獲取請(qǐng)求來源IP的位置信息，判斷請(qǐng)求來源的DC并將請(qǐng)求路由到相應(yīng)的DC的HDFS。如圖9示例所示，若數(shù)據(jù)請(qǐng)求來自DC1, 則Router將數(shù)據(jù)請(qǐng)求重定向到DC1的HDFS集群，來自DC2則定向到DC2的HDFS集群（圖中同種顏色線條標(biāo)識(shí)請(qǐng)求路徑）。

圖9 基于Router的數(shù)據(jù)路由

為了降低跨機(jī)房帶寬的消耗，原則上，我們規(guī)定所有對(duì)數(shù)據(jù)的讀取操作，都只允許在本地機(jī)房（即Client所在機(jī)房）， 否則先拷貝到本地機(jī)房。但特殊情況下，如圖10所示，若Data Replication Service發(fā)生異常短時(shí)間無法修復(fù)或ns長(zhǎng)時(shí)間異常時(shí)，則我們?cè)试S降級(jí)為跨機(jī)房限流讀（副本未ready情況, 超過一定的時(shí)間未在目標(biāo)機(jī)房讀取到數(shù)據(jù)，則降級(jí)），限流部分在后面章節(jié)進(jìn)行詳細(xì)介紹。

圖10 數(shù)據(jù)路由容錯(cuò)

另外 ，由于歷史原因，在我們的生產(chǎn)中存在一種特殊的臨時(shí)庫(kù)，用于管理用戶SQL作業(yè)中的創(chuàng)建的短生命周期的臨時(shí)表（Temporary table，七天自動(dòng)清理），該類臨時(shí)表表名不固定（例如一些ETL作業(yè)會(huì)在臨時(shí)表名上加上日期后綴)，也就造成了表類路徑不固定；針對(duì)該類路徑不固定的情況，無法使用上述鏡像掛載點(diǎn)進(jìn)行管理，因此， 我們引入一種名叫IDC_FOLLOW的多掛載點(diǎn)，用于掛載多個(gè)機(jī)房中的臨時(shí)庫(kù)路徑；當(dāng)讀寫臨時(shí)表時(shí)，會(huì)依據(jù)Client所在的DC選擇DC內(nèi)HDFS NS掛載路徑來存取數(shù)據(jù)，從而解決臨時(shí)表跨機(jī)房流量的問題。

3.4 版本服務(wù)

分布式場(chǎng)景下，通過數(shù)據(jù)復(fù)制方式產(chǎn)生副本，不可避免會(huì)導(dǎo)致一致性問題，因此，多機(jī)房存在數(shù)據(jù)副本時(shí)，除了涉及上述路由選擇問題外，還必須考慮數(shù)據(jù)版本一致性問題，我們通過引入版本服務(wù)（Version）解決該問題；為了簡(jiǎn)化版本服務(wù)設(shè)計(jì), 針對(duì)大數(shù)據(jù)離線場(chǎng)景寫少讀多的特性，我們依據(jù)CAP理論對(duì)鏡像掛載點(diǎn)的實(shí)現(xiàn)做了一定的取舍，規(guī)定了對(duì)主數(shù)據(jù)可以進(jìn)行所有操作，副本數(shù)據(jù)只允許讀/刪操作；在這個(gè)前提下，我們引入了基于HDFS Editlog的版本服務(wù)，如圖11所示，該服務(wù)以觀察者的身份監(jiān)控向HDFS JournalNodes（JN）訂閱路徑的變更行為，并以操作ID（transaction id）來標(biāo)識(shí)數(shù)據(jù)版本；若訂閱的路徑中數(shù)據(jù)發(fā)生了變化，則會(huì)通過editlog傳導(dǎo)到JN，再由JN通知Version進(jìn)行版本更新；因所有對(duì)數(shù)據(jù)的變更操作都會(huì)記錄editlog，因此，不論SQL場(chǎng)景和非SQL場(chǎng)景，只要數(shù)據(jù)存在變化均可被版本服務(wù)捕捉到，從而可以有效保證數(shù)據(jù)的一致性。

圖11 數(shù)據(jù)版本工作流程

上文2.3節(jié)總體流程所描述的提交作業(yè)時(shí)，當(dāng)獲取到作業(yè)預(yù)期的放置機(jī)房后，檢查依賴數(shù)據(jù)是否Ready的工作也包括版本檢查工作；當(dāng)作業(yè)需要副本數(shù)據(jù)時(shí)，會(huì)通過數(shù)據(jù)傳輸服務(wù)檢查已傳輸?shù)臄?shù)據(jù)副本的版本與版本服務(wù)中訂閱的最新版本是否一致，若一致允許作業(yè)提交使用數(shù)據(jù)副本；否則，作業(yè)臨時(shí)阻塞，待傳輸服務(wù)更新副本數(shù)據(jù)后，則允許提交作業(yè)；若超過一定的時(shí)間未在目標(biāo)機(jī)房讀取到數(shù)據(jù)，則降級(jí)為讀取主數(shù)據(jù)。

3.5 限流服務(wù)

我們的場(chǎng)景下跨機(jī)房帶寬有限（約4Tbps），并且和在線服務(wù)、實(shí)時(shí)服務(wù)等對(duì)延遲更敏感的服務(wù)共用帶寬，為防止離線跨機(jī)房流量（特別是計(jì)劃外的跨機(jī)流量）打滿帶寬影響在線業(yè)務(wù)， 我們引入了基于令牌桶的限流服務(wù)。

圖12 令牌桶

令牌桶限流的核心思想為當(dāng)進(jìn)行某操作需要令牌時(shí)，需要從令牌桶中取出相應(yīng)的令牌數(shù)，如果獲取到令牌則繼續(xù)操作，否則阻塞，用完之后不用放回?；谠撍枷胛覀?cè)O(shè)計(jì)了全局中心限流服務(wù)，我們?cè)贖DFS DistributedFileSystem類基礎(chǔ)上，實(shí)現(xiàn)了具有讀寫限流功能的ThrottledDistributedFileSystem，當(dāng)用戶使用該類去讀寫HDFS的文件時(shí)，ThrottledDistributedFileSystem會(huì)根據(jù)RBF返回的LocatedBlock中的client IDC信息和Block IDC信息，判斷此次讀寫流量是否會(huì)跨機(jī)房，如果是會(huì)先嘗試向ThrottleService發(fā)送申請(qǐng)跨機(jī)房帶寬請(qǐng)求（Token），申請(qǐng)到Token后，再進(jìn)行后續(xù)的HDFS讀寫，如果申請(qǐng)的流量用完后，再向ThrottleService申請(qǐng)新的帶寬Token；除利用令牌桶固有的特性外，我們?cè)诹钆仆暗幕A(chǔ)上實(shí)現(xiàn)了隊(duì)列優(yōu)先級(jí)和加權(quán)公平特性，限流服務(wù)的隊(duì)列優(yōu)先級(jí)和調(diào)度系統(tǒng)中的作業(yè)優(yōu)先級(jí)也做一一映射，來保障多租戶情況下重要服務(wù)可以優(yōu)先獲取到Token；在穩(wěn)定性方面，為了降低限流服務(wù)的壓力，我們?cè)O(shè)置每個(gè)Token代表相對(duì)較大的流量單元，來降低Token的獲取次數(shù)過多帶來的性能影響；為防止限流服務(wù)宕機(jī)導(dǎo)致作業(yè)阻塞，我們?cè)黾恿私导?jí)為本地固定帶寬的策略，同時(shí)隨著計(jì)算引擎持續(xù)接入限流服務(wù)，服務(wù)本身的穩(wěn)定性和請(qǐng)求水位成為瓶頸（單機(jī)100K+ qps），我們通過水平擴(kuò)展服務(wù)的方式增強(qiáng)了限流服務(wù)的性能。

3.6 跨機(jī)房流量分析

隨著多機(jī)房項(xiàng)目的逐漸推進(jìn)，跨機(jī)房流量也日漸增長(zhǎng)，高峰時(shí)刻偶爾會(huì)打滿專線帶寬。為了對(duì)跨機(jī)房帶寬流量進(jìn)行有效管控，我們需要了解哪些作業(yè)貢獻(xiàn)了最多的跨機(jī)房流量，從而進(jìn)行針對(duì)性治理。從離線作業(yè)的角度看，網(wǎng)絡(luò)流量來源主要有三塊：

• 從上游讀取數(shù)據(jù)
• 作業(yè)執(zhí)行過程中不同Executor/Task之間shuffle數(shù)據(jù)
• 寫數(shù)據(jù)到下游表

在B站多機(jī)房的場(chǎng)景中，因?yàn)椴捎脝卧軜?gòu)每機(jī)房均存在獨(dú)立Yarn的集群，作業(yè)不會(huì)跨機(jī)房運(yùn)行也就不存在跨機(jī)房Shuffle數(shù)據(jù)的情況，因此只需考慮讀寫HDFS文件過程中產(chǎn)生的跨機(jī)房流量即可，而讀寫HDFS文件產(chǎn)生的跨機(jī)房流量又可以分為計(jì)劃內(nèi)流量和非計(jì)劃內(nèi)流量?jī)纱箢悾?/p>

• 計(jì)劃內(nèi)流量：3.2 小節(jié)所述數(shù)據(jù)復(fù)制服務(wù)進(jìn)行數(shù)據(jù)副本復(fù)制產(chǎn)生的流量，我們稱為計(jì)劃內(nèi)流量， 該部分?jǐn)?shù)據(jù)大概率會(huì)被多次使用
• 非計(jì)劃內(nèi)流量：即非數(shù)據(jù)復(fù)制服務(wù)產(chǎn)生的數(shù)據(jù)流量，單次(或多次)使用，主要來源有以下幾種可能：a. 計(jì)劃內(nèi)的調(diào)度任務(wù)發(fā)生長(zhǎng)時(shí)間跨度的歷史回刷，依賴的數(shù)據(jù)副本已過期銷毀b. （漏遷/錯(cuò)遷/新增等）放置位置不合理的周期性調(diào)度任務(wù)，可以通過優(yōu)化作業(yè)放置消除c. Adhoc查詢，突發(fā)流量， 單次（或多次）使用，臨時(shí)生產(chǎn)需求，無法預(yù)知需要的數(shù)據(jù)，無法預(yù)先進(jìn)行處理

流量分析工具

在實(shí)際生產(chǎn)過程中，非計(jì)劃內(nèi)流量不可避免，為了對(duì)跨機(jī)房流量進(jìn)行有效管控，我們引入了跨機(jī)房流量分析工具，我們?cè)谝娑撕虳N端做了以下改造：

• 引擎端: 在初始化HDFS Client時(shí)將作業(yè)JobId注入到DFSClient的ClientName中
• DataNode: 在DataXceiver中埋點(diǎn)，從ClientName中解析出JobId，并按JobId和client ip網(wǎng)段合并讀寫流量，每30s輸出統(tǒng)計(jì)結(jié)果到流量日志中

我們將每臺(tái)DN上的跨機(jī)房流量日志進(jìn)行實(shí)時(shí)收集，通過Flink匯總到ClickHouse上，然后聚合分析得出每個(gè)時(shí)間段的跨機(jī)房流量作業(yè)Top10，方便對(duì)跨機(jī)房流量進(jìn)行治理（包括重新放置、緊急查殺、作業(yè)優(yōu)化等）。

圖13 流量日志收集鏈路

以下是我們跨機(jī)房流量分析的監(jiān)控面板：

圖14 跨機(jī)房流量分析

Adhoc流量治理&優(yōu)化

對(duì)于Adhoc類型的非計(jì)劃內(nèi)流量，因?yàn)槠潆S機(jī)性，本文所述多機(jī)房體系中“數(shù)據(jù)復(fù)制-作業(yè)放置-數(shù)據(jù)路由”方式不適用；因此，我們采用一些其它的優(yōu)化手段, 比如通過運(yùn)行時(shí)SQL Scan掃描出依賴的數(shù)據(jù)大小、位置信息，以節(jié)省多機(jī)房帶寬為最主要目標(biāo)，結(jié)合集群的實(shí)際負(fù)載情況，決定SQL調(diào)度哪個(gè)機(jī)房，比如：

• 訪問單張表：作業(yè)調(diào)度至數(shù)據(jù)所在機(jī)房
• 訪問多張表多表在同機(jī)房， 作業(yè)調(diào)度至數(shù)據(jù)所在機(jī)房多表在不同機(jī)房， 作業(yè)調(diào)度至數(shù)據(jù)量較大的表所在機(jī)房；較小表限流讀，或者阻塞通知拷貝服務(wù)拷貝

另外， 對(duì)于Presto這種有多源查詢能力的引擎，我們利用其Connector多源查詢功能力將每個(gè)機(jī)房視為一個(gè)Connector，在多表訪問場(chǎng)景中將子查詢下推發(fā)送到遠(yuǎn)端機(jī)房進(jìn)行處理，以減少垮機(jī)房流量帶寬，詳情見 《Persto在B站的實(shí)踐》 5.2節(jié)多機(jī)房架構(gòu)。

04 小結(jié)&展望

本文描述了B站離線多機(jī)房方案，該方案已平穩(wěn)上線運(yùn)行半年以上，遷移數(shù)據(jù)量近300PB，作業(yè)數(shù)占集群所有作業(yè)數(shù)的1/3。從實(shí)踐的結(jié)果來看該方案在很大程度上解決了跨機(jī)房網(wǎng)絡(luò)帶寬不足、穩(wěn)定性差與離線任務(wù)高效產(chǎn)出之間的矛盾。鑒于當(dāng)前部分大數(shù)據(jù)關(guān)鍵組件的單元化進(jìn)程，在抗網(wǎng)絡(luò)連通性風(fēng)險(xiǎn)方面的能力還有較大的提升空間，后續(xù)我們將不斷的推單元化進(jìn)程，進(jìn)一步降低網(wǎng)絡(luò)問題的影響范圍，同時(shí)賦予部分高優(yōu)化作業(yè)“雙活”的能力。

另外隨著新機(jī)房的持續(xù)擴(kuò)容（老機(jī)房無法擴(kuò)容，新增節(jié)點(diǎn)都會(huì)部署在新機(jī)房），我們也需要持續(xù)遷移更多作業(yè)到新機(jī)房，為了提高遷移的推進(jìn)速度，需要盡量減少對(duì)上下游業(yè)務(wù)方的依賴（如：請(qǐng)求業(yè)務(wù)方協(xié)助對(duì)子業(yè)務(wù)進(jìn)行劃分和梳理），因此，我們需要實(shí)現(xiàn)更智能更自動(dòng)化的待遷移數(shù)據(jù)和作業(yè)的自動(dòng)劃分流程，進(jìn)一步強(qiáng)化使用社區(qū)發(fā)現(xiàn)（Community Detection）算法將DAG能自動(dòng)劃分成多個(gè)內(nèi)聚性較高的子集/社區(qū)，按照社區(qū)粒度進(jìn)行遷移的工作。